Alert - Support for authentication when calling receiver endpoint.

connaisseur/alert.py

@@ -48,6 +48,169 @@ def alerting_required(self, event_category: str) -> bool:
         return bool(self.config.get(event_category))
 
 
+class AlertReceiverAuthentication:
+    """
+    Class to store authentication information for securely sending events to the alert receiver.
+    """
+
+    authentication_config: dict = None
+    authentication_scheme: str = None
+
+    class AlertReceiverAuthenticationInterface:
+        def __init__(self, alert_receiver_config: dict, authentication_config_key: str):
+            if authentication_config_key is not None:
+                try:
+                    self.authentication_config = alert_receiver_config[
+                        authentication_config_key
+                    ]
+                except KeyError as err:
+                    raise ConfigurationError(
+                        "No authentication configuration found for dictionary key:"
+                        f"{authentication_config_key}."
+                    ) from err
+
+                self.authentication_scheme = self.authentication_config.get(
+                    "authentication_scheme", self.authentication_scheme
+                )
+                self._validate_authentication_scheme()
+
+        def _validate_authentication_scheme(self) -> None:
+            if not self.authentication_scheme:
+                raise ConfigurationError(
+                    "The authentication scheme cannot be null or empty."
+                )
+
+            if not self.authentication_scheme.isalpha():
+                raise ConfigurationError(
+                    "The authentication scheme must contain only letters."
+                )
+
+        def get_header(self) -> dict:
+            return {}
+
+    class AlertReceiverNoneAuthentication(AlertReceiverAuthenticationInterface):
+        """
+        Placeholder class for AlertReceiver without authentication.
+        """
+
+        def __init__(self, alert_receiver_config: dict):
+            super().__init__(alert_receiver_config, None)
+
+    class AlertReceiverBasicAuthentication(AlertReceiverAuthenticationInterface):
+        """
+        Class to store authentication information for basic authentication type
+        with username and password.
+        """
+
+        username: str
+        password: str
+        authentication_scheme: str = "Basic"
+
+        def __init__(self, alert_receiver_config: dict):
+            super().__init__(alert_receiver_config, "receiver_authentication_basic")
+
+            try:
+                username_env = self.authentication_config["username_env"]
+                password_env = self.authentication_config["password_env"]
+            except KeyError as err:
+                raise ConfigurationError(
+                    "No username_env or password_env configuration found."
+                ) from err
+
+            try:
+                self.username = os.environ[username_env]
+                self.password = os.environ[password_env]
+            except KeyError as err:
+                raise ConfigurationError(
+                    "No username or password found from environmental variables "
+                    f"{username_env} and {password_env}."
+                ) from err
+
+        def get_header(self) -> dict:
+            return {
+                "Authorization": f"{self.authentication_scheme} {self.username}:{self.password}"
+            }
+
+    class AlertReceiverBearerAuthentication(AlertReceiverAuthenticationInterface):
+        """
+        Class to store authentication information for bearer authentication type which uses a token.
+        """
+
+        token: str
+        authentication_scheme: str = "Bearer"  # default is bearer
+
+        def __init__(self, alert_receiver_config: dict):
+            super().__init__(alert_receiver_config, "receiver_authentication_bearer")
+
+            token_env = self.authentication_config.get("token_env")
+            token_file = self.authentication_config.get("token_file")
+
+            if (
+                token_env is None and token_file is None
+            ):  # This should not happen since it is included in the json validation
+                raise ConfigurationError(
+                    "No token_env and token_file configuration found."
+                )
+
+            if (
+                token_env is not None and token_file is not None
+            ):  # This should not happen since it is included in the json validation
+                raise ConfigurationError(
+                    "Both token_env and token_file configuration found. Only one can be given."
+                )
+
+            if token_env is not None:
+                try:
+                    self.token = os.environ[token_env]
+                except KeyError as err:
+                    raise ConfigurationError(
+                        f"No token found from environmental variable {token_env}."
+                    ) from err
+            else:
+                try:
+                    with open(token_file, "r", encoding="utf-8") as token_file_handler:
+                        self.token = token_file_handler.read()
+                except FileNotFoundError as err:
+                    raise ConfigurationError(
+                        f"No token file found at {token_file}."
+                    ) from err
+                except Exception as err:
+                    raise ConfigurationError(
+                        f"An error occurred while loading the token file {token_file}: {str(err)}"
+                    ) from err
+
+        def get_header(self) -> dict:
+            return {"Authorization": f"{self.authentication_scheme} {self.token}"}
+
+    init_map = {
+        "basic": AlertReceiverBasicAuthentication,
+        "bearer": AlertReceiverBearerAuthentication,
+        "none": AlertReceiverNoneAuthentication,
+    }
+
+    _authentication_instance = None
+
+    def __init__(self, alert_receiver_config: dict):
+        self.authentication_type = alert_receiver_config.get(
+            "receiver_authentication_type", "none"
+        )
+        self.__init_authentication_instance(alert_receiver_config)
+
+    def __init_authentication_instance(self, alert_receiver_config: dict):
+        try:
+            self._authentication_instance = self.init_map[self.authentication_type](
+                alert_receiver_config
+            )
+        except KeyError as err:
+            raise ConfigurationError(
+                "No authentication type found. Valid values are "
+                f"{list(AlertReceiverAuthentication.init_map.keys())}"
+            ) from err  # hopefully this never happens
+
+    def get_auth_header(self) -> dict:
+        return self._authentication_instance.get_header()
+
+
 class Alert:
     """
     Class to store image information about an alert as attributes and a sending
@@ -59,6 +222,7 @@ class Alert:
 
     template: str
     receiver_url: str
+    receiver_authentication: AlertReceiverAuthentication
     payload: str
     headers: dict
 
@@ -101,12 +265,13 @@ def __init__(
             "images": images,
         }
         self.receiver_url = receiver_config["receiver_url"]
+        self.receiver_authentication = AlertReceiverAuthentication(receiver_config)
         self.template = receiver_config["template"]
         self.throw_if_alert_sending_fails = receiver_config.get(
             "fail_if_alert_sending_fails", False
         )
         self.payload = self.__construct_payload(receiver_config)
-        self.headers = self.__get_headers(receiver_config)
+        self.headers = self.__get_headers(receiver_config, self.receiver_authentication)
 
     def __construct_payload(self, receiver_config: dict) -> str:
         try:
@@ -153,13 +318,18 @@ def send_alert(self) -> Optional[requests.Response]:
         return response
 
     @staticmethod
-    def __get_headers(receiver_config):
+    def __get_headers(
+        receiver_config: dict, receiver_authentication: AlertReceiverAuthentication
+    ) -> dict:
         headers = {"Content-Type": "application/json"}
         additional_headers = receiver_config.get("custom_headers")
         if additional_headers is not None:
             for header in additional_headers:
                 key, value = header.split(":", 1)
                 headers.update({key.strip(): value.strip()})
+        auth_header = receiver_authentication.get_auth_header()
+        if auth_header:  # not None and not empty
+            headers.update(auth_header)
         return headers
 
 

connaisseur/res/alertconfig_schema.json

@@ -18,6 +18,50 @@
               "receiver_url": {
                 "type": "string"
               },
+              "receiver_authentication_type": {
+                "type": "string",
+                "enum": [
+                  "none",
+                  "basic",
+                  "bearer"
+                ]
+              },
+              "receiver_authentication_basic": {
+                "type": "object",
+                "properties": {
+                  "username_env": {
+                    "type": "string"
+                  },
+                  "password_env": {
+                    "type": "string"
+                  },
+                  "authentication_scheme": {
+                    "type": "string"
+                  }
+                },
+                "required": [
+                  "username_env",
+                  "password_env"
+                ]
+              },
+              "receiver_authentication_bearer": {
+                "type": "object",
+                "oneOf": [
+                  {"required": ["token_file"]},
+                  {"required": ["token_env"]}
+                ],
+                "properties": {
+                  "authentication_scheme": {
+                    "type": "string"
+                  },
+                  "token_file": {
+                    "type": "string"
+                  },
+                  "token_env": {
+                    "type": "string"
+                  }
+                }
+              },
               "priority": {
                 "type": "integer"
               },
@@ -35,6 +79,35 @@
                 "type": "boolean"
               }
             },
+            "anyOf": [
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "basic"
+                  }
+                },
+                "required": [
+                  "receiver_authentication_basic"
+                ]
+              },
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "bearer"
+                  }
+                },
+                "required": [
+                  "receiver_authentication_bearer"
+                ]
+              },
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "none"
+                  }
+                }
+              }
+            ],
             "required": [
               "template",
               "receiver_url"
@@ -60,6 +133,50 @@
               "receiver_url": {
                 "type": "string"
               },
+              "receiver_authentication_type": {
+                "type": "string",
+                "enum": [
+                  "none",
+                  "basic",
+                  "bearer"
+                ]
+              },
+              "receiver_authentication_basic": {
+                "type": "object",
+                "properties": {
+                  "username_env": {
+                    "type": "string"
+                  },
+                  "password_env": {
+                    "type": "string"
+                  },
+                  "authentication_scheme": {
+                    "type": "string"
+                  }
+                },
+                "required": [
+                  "username_env",
+                  "password_env"
+                ]
+              },
+              "receiver_authentication_bearer": {
+                "type": "object",
+                "oneOf": [
+                  {"required": ["token_file"]},
+                  {"required": ["token_env"]}
+                ],
+                "properties": {
+                  "authentication_scheme": {
+                    "type": "string"
+                  },
+                  "token_file": {
+                    "type": "string"
+                  },
+                  "token_env": {
+                    "type": "string"
+                  }
+                }
+              },
               "priority": {
                 "type": "integer"
               },
@@ -77,6 +194,35 @@
                 "type": "boolean"
               }
             },
+            "anyOf": [
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "basic"
+                  }
+                },
+                "required": [
+                  "receiver_authentication_basic"
+                ]
+              },
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "bearer"
+                  }
+                },
+                "required": [
+                  "receiver_authentication_bearer"
+                ]
+              },
+              {
+                "properties": {
+                  "receiver_authentication_type": {
+                    "const": "none"
+                  }
+                }
+              }
+            ],
             "required": [
               "template",
               "receiver_url"