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,

utils/hotfix_utils/InstrumentChecker.py → utils/hotfix_utils/instrument_checker.py

@@ -1,7 +1,7 @@
 """A module for checking the status of an instrument in relation to it's repo."""
 
 import os
-from typing import List, Tuple, Union
+from typing import Any
 
 from ..communication_utils.ssh_access import (
     SSHAccessUtils,
@@ -15,14 +15,22 @@ class InstrumentChecker:
 
     repo_dir = os.environ["REPO_DIR"]
 
-    def __init__(self, hostname: str) -> None:
+    def __init__(
+        self, hostname: str, ssh_username: str, ssh_key_file: str, ssh_passphrase: str
+    ) -> None:
         """Initialize the Instrument object.
 
         Args:
             hostname (str): The hostname of the instrument.
+            ssh_username (str): The ssh username.
+            ssh_key_file (str): The ssh key file.
+            ssh_passphrase (str): The ssh passphrase.
 
         """
         self._hostname = hostname
+        self._ssh_username = ssh_username
+        self._ssh_key_file = ssh_key_file
+        self._ssh_passphrase = ssh_passphrase
 
         self._commits_local_not_on_upstream_enum = None
         self._commits_local_not_on_upstream_messages = None
@@ -43,7 +51,37 @@ def hostname(self) -> str:
         """
         return self._hostname
 
-    def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
+    @property
+    def ssh_username(self) -> str:
+        """Get the ssh username for the instrument.
+
+        Returns:
+            str: The ssh username for the instrument.
+
+        """
+        return self._ssh_username
+
+    @property
+    def ssh_key_file(self) -> str:
+        """Get the ssh key file for the instrument.
+
+        Returns:
+            str: The ssh key file for the instrument.
+
+        """
+        return self._ssh_key_file
+
+    @property
+    def ssh_passphrase(self) -> str:
+        """Get the ssh passphrase for the key.
+
+        Returns:
+            str: The ssh passphrase for the key.
+
+        """
+        return self._ssh_passphrase
+
+    def check_for_uncommitted_changes(self) -> tuple[CHECK, list[Any]]:
         """Check if there are any uncommitted changes on the instrument via SSH.
 
         Args:
@@ -56,8 +94,9 @@ def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
         command = f"cd /d {self.repo_dir} && git status --porcelain"
         ssh_process = SSHAccessUtils.run_ssh_command(
             self.hostname,
-            os.environ["SSH_CREDENTIALS_USR"],
-            os.environ["SSH_CREDENTIALS_PSW"],
+            self.ssh_username,
+            self.ssh_key_file,
+            self.ssh_passphrase,
             command,
         )
 
@@ -67,8 +106,9 @@ def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
         command = f"cd /d {self.repo_dir} && git --no-pager diff --ignore-cr-at-eol"
         ssh_process_diff = SSHAccessUtils.run_ssh_command(
             self.hostname,
-            os.environ["SSH_CREDENTIALS_USR"],
-            os.environ["SSH_CREDENTIALS_PSW"],
+            self.ssh_username,
+            self.ssh_key_file,
+            self.ssh_passphrase,
             command,
         )
 
@@ -81,10 +121,15 @@ def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
                 status_save = status + "\n\n" + ssh_process_diff["output"]
             else:
                 status_save = status
-            JenkinsUtils.save_git_status(self.hostname, status_save, os.environ["WORKSPACE"])
+            JenkinsUtils.save_git_status(
+                self.hostname, str(status_save), os.environ["WORKSPACE"]
+            )
 
             status_stripped = status.strip()
-            if status_stripped != "" and os.environ["SHOW_UNCOMMITTED_CHANGES_MESSAGES"] == "true":
+            if (
+                status_stripped != ""
+                and os.environ["SHOW_UNCOMMITTED_CHANGES_MESSAGES"] == "true"
+            ):
                 return CHECK.TRUE, status_stripped.split("\n")
             elif status_stripped != "":
                 return CHECK.TRUE, []
@@ -96,7 +141,7 @@ def check_for_uncommitted_changes(self) -> Tuple[CHECK, List[any]]:
     def get_parent_epics_branch(
         self,
         hostname: str,
-    ) -> Union[str | bool]:
+    ) -> str | bool:
         """Get the parent branch of the instrument branch.
 
         Args:
@@ -109,8 +154,9 @@ def get_parent_epics_branch(
         command = f"cd /d {self.repo_dir} && git log"
         ssh_process = SSHAccessUtils.run_ssh_command(
             hostname,
-            os.environ["SSH_CREDENTIALS_USR"],
-            os.environ["SSH_CREDENTIALS_PSW"],
+            self.ssh_username,
+            self.ssh_key_file,
+            self.ssh_passphrase,
             command,
         )
         if ssh_process["success"]:
@@ -125,8 +171,8 @@ def git_branch_comparer(
         self,
         hostname: str,
         changes_on: str,
-        subtracted_against: str = None,
-        prefix: str = None,
+        subtracted_against: str | None = None,
+        prefix: str | None = None,
     ) -> CHECK:
         """Get the commit messages between two branches on the instrument.
 
@@ -151,8 +197,9 @@ def git_branch_comparer(
         fetch_command = f"cd /d {self.repo_dir} && git fetch origin"
         ssh_process_fetch = SSHAccessUtils.run_ssh_command(
             hostname,
-            os.environ["SSH_CREDENTIALS_USR"],
-            os.environ["SSH_CREDENTIALS_PSW"],
+            self.ssh_username,
+            self.ssh_key_file,
+            self.ssh_passphrase,
             fetch_command,
         )
 
@@ -172,8 +219,9 @@ def git_branch_comparer(
 
         ssh_process = SSHAccessUtils.run_ssh_command(
             hostname,
-            os.environ["SSH_CREDENTIALS_USR"],
-            os.environ["SSH_CREDENTIALS_PSW"],
+            self.ssh_username,
+            self.ssh_key_file,
+            self.ssh_passphrase,
             command,
         )
 
@@ -203,9 +251,11 @@ def split_git_log(self, git_log: str, prefix: str) -> dict:
 
         Args:
             git_log (str): The git log to split.
+            prefix (str): message prefix to match
 
         Returns:
-            dict: A dictionary with the commit hashes as keys and the commit messages as values.
+            dict: A dictionary with the commit hashes as keys and the
+                  commit messages as values.
 
         """
         commit_dict = {}
@@ -229,7 +279,8 @@ def check_instrument(self) -> dict:
             dict: A dictionary with the result of the checks.
 
         """
-        # Examples of how to use the git_branch_comparer function decided to not be used in this iteration of the check
+        # Examples of how to use the git_branch_comparer function decided to not
+        # be used in this iteration of the check
         # Check if any hotfixes run on the instrument with the prefix "Hotfix:"
         # hotfix_commits_enum, hotfix_commits_messages = git_branch_comparer(
         #     hostname, local_branch, upstream_branch, prefix="Hotfix:")
@@ -245,7 +296,9 @@ def check_instrument(self) -> dict:
         elif os.environ["UPSTREAM_BRANCH_CONFIG"] == "master":
             upstream_branch = "origin/master"
         else:
-            # if the UPSTREAM_BRANCH_CONFIG is not set to any of the above,  set it to the value of the environment variable assuming user wants custom branch
+            # if the UPSTREAM_BRANCH_CONFIG is not set to any of the above,
+            # set it to the value of the environment variable assuming user
+            # wants custom branch
             upstream_branch = os.environ["UPSTREAM_BRANCH_CONFIG"]
 
         # Check if any commits on upstream that are not on the local branch
@@ -282,4 +335,11 @@ def as_string(self) -> str:
             str: The Instrument object as a string.
 
         """
-        return f"Hostname: {self.hostname} - Uncommitted changes: {self.uncommitted_changes_enum} - Commits on local not on upstream: {self.commits_local_not_on_upstream_enum} - Commits on upstream not on local: {self.commits_upstream_not_on_local_enum}"
+        return (
+            f"Hostname: {self.hostname} - "
+            f"Uncommitted changes: {self.uncommitted_changes_enum} "
+            "- Commits on local not on upstream: "
+            f"{self.commits_local_not_on_upstream_enum} "
+            "- Commits on upstream not on local: "
+            f"{self.commits_upstream_not_on_local_enum}"
+        )