Fixes for Debian bookworm

setup.cfg

@@ -1,6 +1,6 @@
 [tool:pytest]
 testpaths = tests
-addopts = -v
+addopts = -v --cov=src --cov-config=setup.cfg --cov-report=term-missing
 
 [coverage:run]
 branch = True

src/lib/command.py

@@ -18,9 +18,13 @@ def checkRemoteHostViaSshProtocol(self, job,
         time.sleep(initial_wait)
         for x in range(retries):
             try:
+                kwargs = {}
+                if job.sshdisabledalgs:
+                    kwargs['disabled_algorithms'] = job.sshdisabledalgs
                 ssh.connect(job.hostname,
                             username=job.sshusername,
-                            key_filename=job.sshprivatekey)
+                            key_filename=job.sshprivatekey,
+                            **kwargs)
                 logger().info(("Successfully connected to host"
                                " via ssh protocol (%s)") % job.hostname)
                 return True
@@ -38,9 +42,13 @@ def executeRemoteCommand(self, job, command):
         ssh = paramiko.SSHClient()
         ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
         try:
+            kwargs = {}
+            if job.sshdisabledalgs:
+                kwargs['disabled_algorithms'] = job.sshdisabledalgs
             ssh.connect(job.hostname,
                         username=job.sshusername,
-                        key_filename=job.sshprivatekey)
+                        key_filename=job.sshprivatekey,
+                        **kwargs)
             logger().info(("Successfully connected to host"
                            " via ssh protocol (%s)") % job.hostname)
 

src/lib/director.py

@@ -23,7 +23,7 @@ def getJobArray(self, jobpath=None):
         jobArray = []
         if jobpath is None:
             directory = config().jobconfigdirectory.rstrip('/')
-            if(os.path.exists(directory)):
+            if (os.path.exists(directory)):
                 os.chdir(directory)
                 for filename in glob.glob("*.job"):
                     j = job(directory + "/" + filename)
@@ -196,10 +196,10 @@ def backupRotate(self, job, moveCurrent=True):
         self._unlinkExpiredBackups(job)
 
         # Rotate backups
-        if(self._rotateBackups(job)):
+        if (self._rotateBackups(job)):
             latest = self._moveCurrentBackup(job)
             if latest:
-                if(self._updateLatestSymlink(job, latest)):
+                if (self._updateLatestSymlink(job, latest)):
                     pass
                 else:
                     logger().error(("Error updating current symlink"
@@ -339,10 +339,10 @@ def _moveLastBackupToCurrentBackup(self, job):
     def getWorkingDirectory(self):
         """Check in which folder we place the backup today"""
         ret = "daily"
-        if(int(datetime.datetime.today().strftime("%w")) ==
+        if (int(datetime.datetime.today().strftime("%w")) ==
            config().weeklybackup):
             ret = "weekly"
-        if(int(datetime.datetime.today().strftime("%d")) ==
+        if (int(datetime.datetime.today().strftime("%d")) ==
            config().monthlybackup):
             ret = "monthly"
         return ret
@@ -383,7 +383,7 @@ def checkWorkingDirectory(self, workingDirectory):
 
     def processBackupStatus(self, job):
         job.backupstatus['hostname'] = job.hostname
-        if(job.ssh):
+        if (job.ssh):
             ssh = 'True'
             job.backupstatus['username'] = job.sshusername
         else:

src/lib/logger.py

@@ -65,7 +65,7 @@ def __init__(self, logfile=None):
             # Create and remember instance
             logger.__instance = logger.__impl()
 
-            if(logfile):  # pragma: no cover
+            if (logfile):  # pragma: no cover
                 logdirectory = os.path.dirname(logfile)
                 if not os.path.exists(logdirectory):
                     os.makedirs(logdirectory)

src/lib/rsync.py

@@ -60,9 +60,13 @@ def checkRemoteHostViaSshProtocol(self, job,
         time.sleep(initial_wait)
         for x in range(retries):
             try:
+                kwargs = {}
+                if job.sshdisabledalgs:
+                    kwargs['disabled_algorithms'] = job.sshdisabledalgs
                 ssh.connect(job.hostname,
                             username=job.sshusername,
-                            key_filename=job.sshprivatekey)
+                            key_filename=job.sshprivatekey,
+                            **kwargs)
                 logger().info(("Successfully connected to host"
                                " via ssh protocol (%s)") % job.hostname)
                 return True
@@ -103,13 +107,13 @@ def executeRsyncViaRsyncProtocol(self, job, latest):
 
         # Link files to the same inodes as last backup to save disk space
         # and boost backup performance
-        if(latest):
+        if (latest):
             latest = "--link-dest=%s" % latest
         else:
             latest = ""
 
         # Generate rsync CLI command and execute it
-        if(include):
+        if (include):
             password = "export RSYNC_PASSWORD=\"%s\"" % job.rsyncpassword
             rsyncCommand = "%s %s %s %s %s" % (
                            config().rsyncpath, options, latest, include, dir)
@@ -139,13 +143,13 @@ def executeRsyncViaSshProtocol(self, job, latest):
 
         # Link files to the same inodes as last backup to save disk space
         # and boost backup performance
-        if(latest):
+        if (latest):
             latest = "--link-dest=%s" % latest
         else:
             latest = ""
 
         # Generate rsync CLI command and execute it
-        if(include):
+        if (include):
             command = "%s %s %s %s %s" % (
                       config().rsyncpath, options, latest, include, directory)
             logger().info("Executing rsync command (%s)" % command)

src/lib/statusemail.py

@@ -116,7 +116,8 @@ def getTextExceptionBody(self, exc):
         return template.render(exc=exc)
 
     def getOverallBackupState(self, jobs):
-        """Overall backup state = 'ok' unless there is at least one failed backup.
+        """Overall backup state = 'ok' unless there is at least one failed
+           backup.
            At the same time reorder the list so errors appear first."""
         ret = "ok"
         good = []

src/models/config.py

@@ -45,7 +45,7 @@ def __init__(self, mainconfigpath=None):
             # Create and remember instance
             config.__instance = config.__impl()
 
-            if(mainconfigpath):  # pragma: no cover
+            if (mainconfigpath):  # pragma: no cover
                 self.mainconfigpath = mainconfigpath
             self.readConfig()
             self.init = False

src/models/job.py

@@ -17,6 +17,7 @@ def __init__(self, filepath=None):
         self.rsyncshare = None
         self.sshusername = None
         self.sshprivatekey = None
+        self.sshdisabledalgs = None
         self.port = None
         self.backupdir = None
         self.speedlimitkb = None
@@ -112,6 +113,13 @@ def readJob(self):
                 self.enabled = False
                 return False
 
+        try:
+            self.sshdisabledalgs = jobconfig['ssh_disabledalgs']
+        except Exception:
+            self.sshdisabledalgs = {}
+            logger().debug(("%s: No or invalid ssh_disabledalgs is set,"
+                            " using default") % self.filepath)
+
         try:
             self.port = jobconfig['port']
         except Exception:

tests/etc/localhost.job

@@ -6,6 +6,10 @@ ssh: False
 ssh_sudo: False
 ssh_username: autorsyncbackup
 ssh_privatekey: /home/autorsyncbackup/.ssh/id_rsa
+ssh_disabledalgs:
+  pubkeys:
+    - rsa-sha2-512
+    - rsa-sha2-256
 port: 10873
 backupdir: /tmp
 speedlimitkb: 10000

tests/etc/ssh-disabledalgs.job

@@ -0,0 +1,15 @@
+hostname: localhost
+ssh: True
+ssh_sudo: False
+ssh_username: autorsyncbackup
+ssh_privatekey: /home/autorsyncbackup/.ssh/id_rsa
+ssh_disabledalgs:
+  pubkeys:
+    - rsa-sha2-512
+    - rsa-sha2-256
+backupdir: /tmp
+include: #formerly fileset
+  - /etc
+exclude:
+  - "*.bak"
+  - ".cache/*"

tests/etc/ssh-no-disabledalgs.job

@@ -0,0 +1,5 @@
+hostname: 'localhost'
+ssh: true
+ssh_sudo: true
+ssh_username: 'autorsyncbackup'
+ssh_privatekey: /home/autorsyncbackup/.ssh/id_rsa

tests/test_command.py

@@ -9,7 +9,10 @@
 
 
 def test_checkRemoteHostViaSshProtocol(monkeypatch):
-    def mock_connect(self, hostname, username=None, key_filename=None):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
         return True
 
     monkeypatch.setattr(paramiko.SSHClient, 'connect', mock_connect)
@@ -28,10 +31,36 @@ def mock_connect(self, hostname, username=None, key_filename=None):
     assert ret is True
 
 
+def test_checkRemoteHostViaSshProtocol_no_disabledalgs(monkeypatch):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
+        return True
+
+    monkeypatch.setattr(paramiko.SSHClient, 'connect', mock_connect)
+
+    path = os.path.join(
+                         os.path.dirname(__file__),
+                         'etc/ssh-no-disabledalgs.job',
+                       )
+
+    j = job(path)
+
+    cmd = command()
+
+    ret = cmd.checkRemoteHostViaSshProtocol(j)
+
+    assert ret is True
+
+
 def test_checkRemoteHostViaSshProtocol_exception(monkeypatch, caplog):
     logger().debuglevel = 3
 
-    def mock_connect(self, hostname, username=None, key_filename=None):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
         raise IOError('Mock connection failed')
 
     monkeypatch.setattr(paramiko.SSHClient, 'connect', mock_connect)
@@ -53,7 +82,10 @@ def mock_connect(self, hostname, username=None, key_filename=None):
 
 
 def test_executeRemoteCommand(monkeypatch):
-    def mock_connect(self, hostname, username=None, key_filename=None):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
         return True
 
     def mock_exec_command(self, command):
@@ -81,10 +113,45 @@ def mock_exec_command(self, command):
     assert 'Mock STDOUT' in stdout
 
 
+def test_executeRemoteCommand_no_disabledalgs(monkeypatch):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
+        return True
+
+    def mock_exec_command(self, command):
+        stdin = io.StringIO('')
+        stdout = io.StringIO('Mock STDOUT\n0')
+        stderr = io.StringIO('')
+
+        return stdin, stdout, stderr
+
+    monkeypatch.setattr(paramiko.SSHClient, 'connect', mock_connect)
+    monkeypatch.setattr(paramiko.SSHClient, 'exec_command', mock_exec_command)
+
+    path = os.path.join(
+                         os.path.dirname(__file__),
+                         'etc/ssh-no-disabledalgs.job',
+                       )
+
+    j = job(path)
+
+    cmd = command()
+
+    (status, stdout, stderr) = cmd.executeRemoteCommand(j, 'uptime')
+
+    assert status == 0
+    assert 'Mock STDOUT' in stdout
+
+
 def test_executeRemoteCommand_exception(monkeypatch, caplog):
     logger().debuglevel = 3
 
-    def mock_connect(self, hostname, username=None, key_filename=None):
+    def mock_connect(
+        self, hostname,
+        username=None, key_filename=None, **kwargs,
+    ):
         raise IOError('Mock connection failed')
 
     monkeypatch.setattr(paramiko.SSHClient, 'connect', mock_connect)