initial commit

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+secrets/

gotify.py

@@ -0,0 +1,24 @@
+import requests
+
+
+class Gotify:
+    def __init__(self, token: str):
+        self._token = token
+        self._url = f"https://gotify.lennartalff.net/message?token={self._token}"
+
+    def send(self, priority: int, title: str, text: str):
+        return requests.post(
+            self._url,
+            json={
+                "message": text,
+                "priority": priority,
+                "title": title,
+            },
+        )
+
+    def send_info(self, title="", text=""):
+        return self.send(priority=5, title=title, text=text)
+
+    def send_error(self, title="", text=""):
+        return self.send(priority=10, title=title, text=text)
+

nextcloud.json.sample

@@ -0,0 +1,16 @@
+// vim: ft=jsonc
+// put this file into the secrets subdir
+{
+    "BORG_RSH": "ssh -i path_to_ssh_key_file",
+    "BORG_PASSPHRASE": "somepassphrase",
+    "GOTIFY_TOKEN": "the_gotify_token_for_logging",
+    "BACKUP_DOCKER_DIR": "dir of the docker compose file",
+    "BACKUP_DIRS": ["list of dirs to backup"],
+    "REPO_SUBDIR": "nextcloud",
+    "BACKUP_USER": "username of the remote repo serving borg",
+    "EXCLUDE_DIRS": ["list of directories to exclude from the backup"],
+    "TIME_FORMAT": "utcnow:%Y-%m-%d_%H:%M:%S",
+    "MYSQL_DB": "database name",
+    "MYSQL_USER": "database user required for dumping the database",
+    "MYSQL_PASSWORD": "password required for dumping the database "
+}

nextcloud_backup

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+
+import gotify
+import json
+import subprocess
+import os
+from pathlib import Path
+
+
+def read_config():
+    source_path = Path(__file__).resolve()
+    nextcloud_secret = source_path.parent / "secrets/nextcloud.json"
+    with open(nextcloud_secret, "r") as f:
+        config = json.load(f)
+    return config
+
+
+class BackupManager:
+    def __init__(self):
+        self._config = read_config()
+        self._gotify = gotify.Gotify(self._config["GOTIFY_TOKEN"])
+
+    def enable_maintenance(self):
+        cmd = "docker compose exec -i --user 1000:1000 app /var/www/html/occ maintenance:mode --on"
+        try:
+            result = subprocess.run(
+                cmd, shell=True, text=True, check=True, capture_output=True
+            )
+        except subprocess.CalledProcessError as e:
+            self._gotify.send_error(
+                "❗💀❗ Enabling maintenace failed",
+                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
+            )
+            return False
+        if "Maintenance mode already enabled" in result.stdout:
+            self._gotify.send_info(
+                "❗ Maintenance unexpectedly enabled",
+                (
+                    "Maintenance mode was already enabled. "
+                    "Did not expect that. Will continue."
+                ),
+            )
+        return True
+
+    def disable_maintenance(self):
+        cmd = "docker compose exec -i --user 1000:1000 app /var/www/html/occ maintenance:mode --off"
+        try:
+            result = subprocess.run(
+                cmd, shell=True, text=True, check=True, capture_output=True
+            )
+        except subprocess.CalledProcessError as e:
+            self._gotify.send_error(
+                "❗💀❗ Disabling maintenace failed",
+                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
+            )
+            return False
+        if "Maintenance mode already disabled" in result.stdout:
+            self._gotify.send_info(
+                "❗ Maintenance mode unexpectedly already disabled",
+                (
+                    "Maintenance mode was already disabled. "
+                    "Did not expect that. Will continue."
+                ),
+            )
+        return True
+
+    def dump_database(self):
+        password = self._config["MYSQL_PASSWORD"]
+        user = self._config["MYSQL_USER"]
+        db = self._config["MYSQL_DB"]
+
+        cmd = f"docker compose exec -i --user 1000:1000 db mariadb-dump --single-transaction --default-character-set=utf8mb4 -h localhost -u {user} --password={password} {db} > db/nextcloud.sql"
+        try:
+            result = subprocess.run(
+                cmd,
+                shell=True,
+                check=True,
+                text=True,
+                capture_output=True,
+            )
+        except subprocess.CalledProcessError as e:
+            self._gotify.send_error(
+                "❗💀❗ Dumping Database failed",
+                f"stdout:\n{e.stdout}\nstderr:\n{e.stderr}",
+            )
+            return False
+        text = "\n".join([result.stdout, result.stderr])
+        self._gotify.send_info("✅ Database dumped", f"Result:\n{text}")
+        return True
+
+    def borg_backup(self):
+        backup_dirs = " ".join(self._config["BACKUP_DIRS"])
+        exclude_dirs = " ".join(self._config["EXCLUDE_DIRS"])
+        repo_subdir = self._config["REPO_SUBDIR"]
+        time_format = self._config["TIME_FORMAT"]
+        backup_user = self._config["BACKUP_USER"]
+        hostname = os.uname().nodename
+        borg_env = os.environ.copy()
+        borg_env["BORG_RSH"] = "ssh -i /home/lennartalff/.ssh/borg.ed25519"
+        borg_env["BORG_PASSPHRASE"] = self._config["BORG_PASSPHRASE"]
+        repo = f"ssh://{backup_user}@{backup_user}.your-storagebox.de:23/./backups/{hostname}/{repo_subdir}::{{{time_format}}}"
+        cmd = f"borg create -v --stats {repo} {backup_dirs} --exclude {exclude_dirs}"
+        try:
+            result = subprocess.run(
+                cmd,
+                shell=True,
+                check=True,
+                text=True,
+                capture_output=True,
+                env=borg_env,
+            )
+        except subprocess.CalledProcessError as e:
+            self._gotify.send_error(
+                title="❗💀❗ Backup failed!",
+                text=f"stdout: \n{e.stdout}\nsterr: \n{e.stderr}",
+            )
+            return False
+        text = "\n".join([result.stdout, result.stderr])
+        self._gotify.send_info("✅ Backup completed", f"Result:\n{text}\n")
+        return True
+
+
+def main():
+    config = read_config()
+    os.chdir(config["BACKUP_DOCKER_DIR"])
+    backup_manager = BackupManager()
+    if not backup_manager.enable_maintenance():
+        backup_manager.disable_maintenance()
+        exit(1)
+    if not backup_manager.dump_database():
+        backup_manager.disable_maintenance()
+        exit(1)
+    if not backup_manager.borg_backup():
+        backup_manager.disable_maintenance()
+        exit(1)
+    if not backup_manager.disable_maintenance():
+        exit(1)
+    exit(0)
+
+
+if __name__ == "__main__":
+    main()