Fix ssh access

Jenkinsfile_config_dir

@@ -19,7 +19,6 @@ pipeline {
         skipDefaultCheckout(true)
     }
     environment {
-        SSH_CREDENTIALS = credentials('SSH')
         TEST_INSTRUMENT_LIST = "${TEST_INSTRUMENT_LIST}"
         USE_TEST_INSTRUMENT_LIST = "${USE_TEST_INSTRUMENT_LIST}"
         DEBUG_MODE = "${DEBUG_MODE}"
@@ -42,11 +41,18 @@ pipeline {
         stage('Check Instrument has any Hotfixes and then any uncommitteed changes') {
             steps {
                 echo 'Check Instrument has any commits or any uncommitteed changes'
-                timeout(time: 1, unit: 'HOURS') {
-                    bat '''
-                        call get_python.bat
-                        python -u hotfix_checker.py
-                    '''
+                withCredentials([sshUserPrivateKey(credentialsId: '9f81ea0c-9740-4e2e-b58a-46d426645acb',
+                                                   usernameVariable: 'SSH_CREDENTIALS_USER',
+                                                   passphraseVariable: 'SSH_CREDENTIALS_PASSPHRASE',                                                   
+                                                   keyFileVariable: 'SSH_CREDENTIALS_KEY_FILE')]) {
+                    timeout(time: 1, unit: 'HOURS') {
+                        bat '''
+                            @echo off
+                            setlocal
+                            call get_python.bat
+                            python -u hotfix_checker.py
+                        '''
+                    }
                 }
             }
         }

Jenkinsfile_epics_dir

@@ -20,7 +20,6 @@ pipeline {
     }
 
     environment {
-        SSH_CREDENTIALS = credentials('SSH')
         TEST_INSTRUMENT_LIST = "${TEST_INSTRUMENT_LIST}"
         USE_TEST_INSTRUMENT_LIST = "${USE_TEST_INSTRUMENT_LIST}"
         DEBUG_MODE = "${DEBUG_MODE}"
@@ -43,11 +42,18 @@ pipeline {
         stage('Check Instrument has any Hotfixes and then any uncommitteed changes') {
             steps {
                 echo 'Check Instrument has any Hotfixes and then any uncommitteed changes'
-                timeout(time: 1, unit: 'HOURS') {
-                    bat '''
-                        call get_python.bat
-                        python -u hotfix_checker.py
-                    '''
+                withCredentials([sshUserPrivateKey(credentialsId: '9f81ea0c-9740-4e2e-b58a-46d426645acb',
+                                                   usernameVariable: 'SSH_CREDENTIALS_USER',
+                                                   passphraseVariable: 'SSH_CREDENTIALS_PASSPHRASE',                                                   
+                                                   keyFileVariable: 'SSH_CREDENTIALS_KEY_FILE')]) {
+                    timeout(time: 1, unit: 'HOURS') {
+                        bat '''
+                            @echo off
+                            setlocal
+                            call get_python.bat
+                            python -u hotfix_checker.py
+                        '''
+                    }
                 }
             }
         }

hotfix_checker.py

@@ -1,4 +1,8 @@
-"""Creates a RepoChecker object and calls the check_instruments method to check for changes in the instruments repository."""
+"""Checks a repository.
+
+Creates a RepoChecker object and calls the check_instruments method
+to check for changes in the instruments repository.
+"""
 
 import os
 
@@ -7,20 +11,24 @@
 # Load environment variables from .env file
 load_dotenv(find_dotenv())
 
-# importing here so it doesn't set variables that are populated from env vars before actually having the env vars loaded
+# importing here so it doesn't set variables that are populated from env vars
+# before actually having the env vars loaded
 # needed for when running locally to get the contents of a .env fil
 # Jenkins will have the env vars set in the pipeline
-from utils.hotfix_utils.RepoChecker import RepoChecker
+from utils.hotfix_utils.repo_checker import (
+    RepoChecker,  # see above comments
+)
 
 if __name__ == "__main__":
     if os.environ["DEBUG_MODE"] == "true":
         print("INFO: Running in debug mode")
         print(f"INFO: REPO_DIR: {os.environ['REPO_DIR']}")
         print(f"INFO: UPSTREAM_BRANCH: {os.environ['UPSTREAM_BRANCH_CONFIG']}")
         print(f"INFO: ARTEFACT_DIR: {os.environ['WORKSPACE']}")
-        print(f"INFO: USE_TEST_INSTRUMENT_LIST: {os.environ['USE_TEST_INSTRUMENT_LIST']}")
+        print(
+            f"INFO: USE_TEST_INSTRUMENT_LIST: {os.environ['USE_TEST_INSTRUMENT_LIST']}"
+        )
         print(f"INFO: TEST_INSTRUMENT_LIST: {os.environ['TEST_INSTRUMENT_LIST']}")
         print(f"INFO: DEBUG_MODE: {os.environ['DEBUG_MODE']}")
-
     repo_checker = RepoChecker()
     repo_checker.check_instruments()

utils/communication_utils/ssh_access.py

@@ -1,28 +1,28 @@
-"""This module provides utilities for SSH access."""
-
-from typing import Dict
+"""Module provides utilities for SSH access."""
 
 import paramiko
 
 SSH_PORT = 22
 
 
-class SSHAccessUtils(object):
+class SSHAccessUtils:
     """Class containing utility methods for SSH access."""
 
     @staticmethod
     def run_ssh_command(
         host: str,
         username: str,
-        password: str,
+        key_file: str,
+        passphrase: str,
         command: str,
-    ) -> Dict[str, bool | str]:
+    ) -> dict[str, bool | str]:
         """Run a command on a remote host using SSH.
 
         Args:
             host (str): The hostname to connect to.
             username (str): The username to use to connect.
-            password (str): The password to use to connect.
+            key_file (str): The ssh key file to use to connect.
+            passphrase (str): The ssh key passphrase to use.
             command (str): The command to run on the remote host.
 
         Returns:
@@ -36,15 +36,16 @@ def run_ssh_command(
                 host,
                 port=SSH_PORT,
                 username=username,
-                password=password,
+                key_filename=key_file,
+                passphrase=passphrase,
             )
             (
-                stdin,
+                _stdin,
                 stdout,
                 stderr,
             ) = client.exec_command(command)
-            output = stdout.read().decode("utf-8")
-            error = stderr.read().decode("utf-8")
+            output = stdout.read().decode("utf-8", errors="backslashreplace")
+            error = stderr.read().decode("utf-8", errors="backslashreplace")
             client.close()
             if error:
                 return {
@@ -56,7 +57,7 @@ def run_ssh_command(
                     "success": True,
                     "output": output,
                 }
-        except Exception as e:
+        except Exception as e:  # noqa: BLE001
             print(str(e))
             return {
                 "success": False,