Alert - Support for authentication when calling receiver endpoint. #560
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -48,6 +48,166 @@ 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. | ||
| """ | ||
|
|
||
| class AlertReceiverBasicAuthentication: | ||
| """ | ||
| Class to store authentication information for basic authentication type with username and password. | ||
| """ | ||
| username: str | ||
| password: str | ||
| authentication_type: str | ||
| authorization_prefix: str = "Basic" | ||
|
|
||
| def __init__(self, alert_receiver_config: dict): | ||
| basic_authentication_config = alert_receiver_config.get( | ||
| "receiver_authentication_basic", None | ||
| ) | ||
|
|
||
| if ( | ||
| basic_authentication_config is None | ||
| ): # TODO maybe remove this check since it is included in the json validation? | ||
|
|
||
| raise ConfigurationError("No basic authentication configuration found.") | ||
|
|
||
| username_env = basic_authentication_config.get("username_env", None) | ||
| password_env = basic_authentication_config.get("password_env", None) | ||
|
|
||
| if ( | ||
| username_env is None or password_env is None | ||
| ): # TODO maybe remove this check since it is included in the json validation? | ||
| raise ConfigurationError( | ||
| "No username_env or password_env configuration found." | ||
| ) | ||
|
|
||
| self.username = os.environ.get(username_env, None) | ||
| self.password = os.environ.get(password_env, None) | ||
|
|
||
| if self.username is None or self.password is None: | ||
| raise ConfigurationError( | ||
| f"No username or password found from environmental variables {username_env} and {password_env}." | ||
| ) | ||
|
|
||
| self.authorization_prefix = basic_authentication_config.get( | ||
| "authorization_prefix", "Basic" | ||
|
There was a problem hiding this comment. Should the default be |
||
| ) | ||
| # TODO maybe validate authorization prefix | ||
|
There was a problem hiding this comment. I cannot find an actual validation rule. |
||
|
|
||
| def get_header(self) -> dict: | ||
| return { | ||
| "Authorization": f"{self.authorization_prefix} {self.username}:{self.password}" | ||
| } | ||
|
|
||
| class AlertReceiverBearerAuthentication: | ||
| """ | ||
| Class to store authentication information for bearer authentication type which uses a token. | ||
| """ | ||
| token: str | ||
| authorization_prefix: str = "Bearer" # default is bearer | ||
|
|
||
| def __init__(self, alert_receiver_config: dict): | ||
| bearer_authentication_config = alert_receiver_config.get( | ||
| "receiver_authentication_bearer", None | ||
| ) | ||
|
|
||
| if ( | ||
| bearer_authentication_config is None | ||
| ): # TODO maybe remove this check since it is included in the json validation? | ||
| raise ConfigurationError( | ||
| "No bearer authentication configuration found." | ||
| ) | ||
|
There was a problem hiding this comment. This is boilerplate (together with assigning Suggestion: make a base class, introduce a variable for the lookup key (e.g. |
||
|
|
||
| token_env = bearer_authentication_config.get("token_env", None) | ||
| token_file = bearer_authentication_config.get("token_file", None) | ||
|
|
||
| if ( | ||
| token_env is None and token_file is None | ||
| ): # TODO maybe remove this check 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 | ||
| ): # TODO maybe remove this check since it is included in the json validation? | ||
| raise ConfigurationError( | ||
| "Both token_env and token_file configuration found. Only one is required." | ||
|
|
||
| ) | ||
|
|
||
| if token_env is not None: | ||
| self.token = os.environ.get(token_env, None) | ||
|
|
||
| if self.token is None: | ||
| raise ConfigurationError( | ||
| f"No token found from environmental variable {token_env}." | ||
| ) | ||
| else: | ||
| try: | ||
| with open(token_file, "r") as token_file: | ||
| self.token = token_file.read() | ||
| except FileNotFoundError: | ||
| raise ConfigurationError(f"No token file found at {token_file}.") | ||
| except Exception as err: | ||
| raise ConfigurationError( | ||
| f"An error occurred while loading the token file {token_file}: {str(err)}" | ||
| ) | ||
|
|
||
| self.authorization_prefix = bearer_authentication_config.get( | ||
| "authorization_prefix", "Bearer" | ||
| ) | ||
| # TODO maybe validate authorization prefix | ||
|
|
||
| def get_header(self) -> dict: | ||
| return {"Authorization": f"{self.authorization_prefix} {self.token}"} | ||
|
|
||
| def __init__(self, alert_receiver_config: dict): | ||
| self.authentication_type = alert_receiver_config.get( | ||
| "receiver_authentication_type", "none" | ||
| ) | ||
|
|
||
| if self.is_basic(): | ||
| self.__init_basic_authentication(alert_receiver_config) | ||
| elif self.is_bearer(): | ||
| self.__init_bearer_authentication(alert_receiver_config) | ||
|
There was a problem hiding this comment. init_map = {
"basic": AlertReceiverAuthentication.AlertReceiverBasicAuthentication,
"bearer": AlertReceiverAuthentication.AlertReceiverBearerAuthentication,
}
try:
self.auth_instance = init_map[self.authentication_type](alert_receiver_config)
except AttributeError:
....Then, it |
||
|
|
||
| def is_basic(self): | ||
| return self.authentication_type == "basic" | ||
|
|
||
| def is_bearer(self): | ||
| return self.authentication_type == "bearer" | ||
|
|
||
| def is_none(self): | ||
| return self.authentication_type == "none" | ||
|
|
||
| def __init_bearer_authentication(self, alert_receiver_config: dict): | ||
| self.bearer_authentication = ( | ||
| AlertReceiverAuthentication.AlertReceiverBearerAuthentication( | ||
| alert_receiver_config | ||
| ) | ||
| ) | ||
|
|
||
| def __init_basic_authentication(self, alert_receiver_config: dict): | ||
| self.basic_authentication = ( | ||
| AlertReceiverAuthentication.AlertReceiverBasicAuthentication( | ||
| alert_receiver_config | ||
| ) | ||
| ) | ||
|
|
||
| def get_auth_header(self) -> dict: | ||
| if self.is_basic(): | ||
| return self.basic_authentication.get_header() | ||
| elif self.is_bearer(): | ||
| return self.bearer_authentication.get_header() | ||
| elif self.is_none(): | ||
| return {} | ||
| else: | ||
| raise ConfigurationError( | ||
| "No authentication type found." | ||
| ) # hopefully this never happens | ||
|
|
||
|
|
||
| class Alert: | ||
| """ | ||
| Class to store image information about an alert as attributes and a sending | ||
|
|
@@ -59,6 +219,7 @@ class Alert: | |
|
|
||
| template: str | ||
| receiver_url: str | ||
| receiver_authentication: AlertReceiverAuthentication | ||
| payload: str | ||
| headers: dict | ||
|
|
||
|
|
@@ -101,12 +262,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.receiver_authentication) | ||
|
|
||
| def __construct_payload(self, receiver_config: dict) -> str: | ||
| try: | ||
|
|
@@ -153,13 +315,18 @@ def send_alert(self) -> Optional[requests.Response]: | |
| return response | ||
|
|
||
| @staticmethod | ||
| 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 | ||
|
|
||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Noneis the default fordict.get. Same for other occurrences.