#99 add aws secrets manager plugin (#102)

* #99 - Adding support to AWS Secrets Manager as a token provider

* #99 - Removing unnecessary comments

* #99 - Adding boto3 to setup.py

* #99 - Adding secret_key parameter

* #99 - Covering unhappy path when secret_key is incorrect

README.md

@@ -67,6 +67,10 @@ cachet:
  token:
  - type: ENVIRONMENT_VARIABLE
  value: CACHET_TOKEN
+ - type: AWS_SECRETS_MANAGER
+ secret_name: cachethq
+ secret_key: token
+ region: us-west-2
  - type: TOKEN
  value: my_token
 webhooks:
@@ -116,6 +120,12 @@ messages:
  . (since 0.6.10) *mandatory*
  - **ENVIRONMENT_VARIABLE**, it will read the token from the specified environment variable.
  - **TOKEN**, it's a string and it will be read directly from the configuration.
+ - **AWS_SECRETS_MANAGER**, it will attempt reading the token from
+ [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/). It requires setting up the AWS credentials
+ into the docker container. More instructions below. It takes these parameters:
+ - **secret_name**, the name of the secret.
+ - **secret_key**, the key under which the token is stored.
+ - **region**, the AWS region.
 - **webhooks**, generic webhooks to be notified about incident updates
  - **url**, webhook URL, will be interpolated
  - **params**, POST parameters, will be interpolated
@@ -146,6 +156,14 @@ Following parameters are available in webhook interpolation
 | `{title}` | Event title, includes endpoint name and short status |
 | `{message}` | Event message, same as sent to Cachet |
 
+### AWS Secrets Manager
+This tools can integrate with AWS Secrets Manager, where the token is fetched directly from the service. In order to
+ get this functionality working, you will need to setup the AWS credentials into the container. The easiest way would
+ be setting the environment variables:
+```bash
+$ docker run --rm -it -e AWS_ACCESS_KEY_ID=xyz -e AWS_SECRET_ACCESS_KEY=aaa -v "$PWD"/my_config.yml:/usr/src/app/config/config.yml:ro mtakaki/cachet-url-monitor
+```
+
 ## Setting up
 
 The application should be installed using **virtualenv**, through the following command:

cachet_url_monitor/plugins/token_provider.py

@@ -1,8 +1,11 @@
+import json
+import os
 from typing import Any
 from typing import Dict
 from typing import Optional
 
-import os
+from boto3.session import Session
+from botocore.exceptions import ClientError
 
 
 class TokenProvider:
@@ -31,9 +34,49 @@ def get_token(self) -> Optional[str]:
  return self.token
 
 
+class AwsSecretsManagerTokenRetrievalException(Exception):
+ def __init__(self, message):
+ self.message = message
+
+ def __repr__(self):
+ return self.message
+
+
+class AwsSecretsManagerTokenProvider(TokenProvider):
+ def __init__(self, config_data: Dict[str, Any]):
+ self.secret_name = config_data["secret_name"]
+ self.region = config_data["region"]
+ self.secret_key = config_data["secret_key"]
+
+ def get_token(self) -> Optional[str]:
+ session = Session()
+ client = session.client(service_name="secretsmanager", region_name=self.region)
+ try:
+ get_secret_value_response = client.get_secret_value(SecretId=self.secret_name)
+ except ClientError as e:
+ if e.response["Error"]["Code"] == "ResourceNotFoundException":
+ raise AwsSecretsManagerTokenRetrievalException(f"The requested secret {self.secret_name} was not found")
+ elif e.response["Error"]["Code"] == "InvalidRequestException":
+ raise AwsSecretsManagerTokenRetrievalException("The request was invalid")
+ elif e.response["Error"]["Code"] == "InvalidParameterException":
+ raise AwsSecretsManagerTokenRetrievalException("The request had invalid params")
+ else:
+ if "SecretString" in get_secret_value_response:
+ secret = json.loads(get_secret_value_response["SecretString"])
+ try:
+ return secret[self.secret_key]
+ except KeyError:
+ raise AwsSecretsManagerTokenRetrievalException(f"Invalid secret_key parameter: {self.secret_key}")
+ else:
+ raise AwsSecretsManagerTokenRetrievalException(
+ "Invalid secret format. It should be a SecretString, instead of binary."
+ )
+
+
 TYPE_NAME_TO_CLASS: Dict[str, TokenProvider] = {
  "ENVIRONMENT_VARIABLE": EnvironmentVariableTokenProvider,
  "TOKEN": ConfigurationFileTokenProvider,
+ "AWS_SECRETS_MANAGER": AwsSecretsManagerTokenProvider,
 }
 
 

requirements.txt

@@ -1,3 +1,4 @@
+boto3==1.13.12
 PyYAML==5.3
 requests==2.22.0
 Click==7.0

setup.py

@@ -10,7 +10,7 @@
  url="https://github.com/mtakaki/cachet-url-monitor",
  packages=find_packages(),
  license="MIT",
- requires=["requests", "PyYAML", "Click"],
+ requires=["requests", "PyYAML", "Click", "boto3"],
  setup_requires=["pytest-runner"],
  tests_require=["pytest", "requests-mock"],
 )

tests/plugins/test_token_provider.py

@@ -4,10 +4,25 @@
 
 from cachet_url_monitor.plugins.token_provider import get_token
 from cachet_url_monitor.plugins.token_provider import get_token_provider_by_name
+from cachet_url_monitor.plugins.token_provider import AwsSecretsManagerTokenProvider
 from cachet_url_monitor.plugins.token_provider import ConfigurationFileTokenProvider
 from cachet_url_monitor.plugins.token_provider import EnvironmentVariableTokenProvider
 from cachet_url_monitor.plugins.token_provider import InvalidTokenProviderTypeException
 from cachet_url_monitor.plugins.token_provider import TokenNotFoundException
+from cachet_url_monitor.plugins.token_provider import AwsSecretsManagerTokenRetrievalException
+
+from botocore.exceptions import ClientError
+
+
+@pytest.fixture()
+def mock_boto3():
+ with mock.patch("cachet_url_monitor.plugins.token_provider.Session") as _mock_session:
+ mock_session = mock.Mock()
+ _mock_session.return_value = mock_session
+
+ mock_client = mock.Mock()
+ mock_session.client.return_value = mock_client
+ yield mock_client
 
 
 def test_configuration_file_token_provider():
@@ -31,6 +46,10 @@ def test_get_token_provider_by_name_environment_variable_type():
  assert get_token_provider_by_name("ENVIRONMENT_VARIABLE") == EnvironmentVariableTokenProvider
 
 
+def test_get_token_provider_by_name_aws_secrets_manager_type():
+ assert get_token_provider_by_name("AWS_SECRETS_MANAGER") == AwsSecretsManagerTokenProvider
+
+
 def test_get_token_provider_by_name_invalid_type():
  with pytest.raises(InvalidTokenProviderTypeException) as exception_info:
  get_token_provider_by_name("WRONG")
@@ -65,3 +84,70 @@ def test_get_token_no_token_found(mock_os):
 def test_get_token_string_configuration():
  token = get_token("my_token")
  assert token == "my_token"
+
+
+def test_get_aws_secrets_manager(mock_boto3):
+ mock_boto3.get_secret_value.return_value = {"SecretString": '{"token": "my_token"}'}
+ token = get_token(
+ [{"secret_name": "hq_token", "type": "AWS_SECRETS_MANAGER", "region": "us-west-2", "secret_key": "token"}]
+ )
+ assert token == "my_token"
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")
+
+
+def test_get_aws_secrets_manager_incorrect_secret_key(mock_boto3):
+ mock_boto3.get_secret_value.return_value = {"SecretString": '{"token": "my_token"}'}
+ with pytest.raises(AwsSecretsManagerTokenRetrievalException):
+ get_token(
+ [
+ {
+ "secret_name": "hq_token",
+ "type": "AWS_SECRETS_MANAGER",
+ "region": "us-west-2",
+ "secret_key": "wrong_key",
+ }
+ ]
+ )
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")
+
+
+def test_get_aws_secrets_manager_binary_secret(mock_boto3):
+ mock_boto3.get_secret_value.return_value = {"binary": "it_will_fail"}
+ with pytest.raises(AwsSecretsManagerTokenRetrievalException):
+ get_token(
+ [{"secret_name": "hq_token", "type": "AWS_SECRETS_MANAGER", "region": "us-west-2", "secret_key": "token"}]
+ )
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")
+
+
+def test_get_aws_secrets_manager_resource_not_found_exception(mock_boto3):
+ mock_boto3.get_secret_value.side_effect = ClientError(
+ error_response={"Error": {"Code": "ResourceNotFoundException"}}, operation_name="get_secret_value"
+ )
+ with pytest.raises(AwsSecretsManagerTokenRetrievalException):
+ get_token(
+ [{"secret_name": "hq_token", "type": "AWS_SECRETS_MANAGER", "region": "us-west-2", "secret_key": "token"}]
+ )
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")
+
+
+def test_get_aws_secrets_manager_invalid_request_exception(mock_boto3):
+ mock_boto3.get_secret_value.side_effect = ClientError(
+ error_response={"Error": {"Code": "InvalidRequestException"}}, operation_name="get_secret_value"
+ )
+ with pytest.raises(AwsSecretsManagerTokenRetrievalException):
+ get_token(
+ [{"secret_name": "hq_token", "type": "AWS_SECRETS_MANAGER", "region": "us-west-2", "secret_key": "token"}]
+ )
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")
+
+
+def test_get_aws_secrets_manager_invalid_parameter_exception(mock_boto3):
+ mock_boto3.get_secret_value.side_effect = ClientError(
+ error_response={"Error": {"Code": "InvalidParameterException"}}, operation_name="get_secret_value"
+ )
+ with pytest.raises(AwsSecretsManagerTokenRetrievalException):
+ get_token(
+ [{"secret_name": "hq_token", "type": "AWS_SECRETS_MANAGER", "region": "us-west-2", "secret_key": "token"}]
+ )
+ mock_boto3.get_secret_value.assert_called_with(SecretId="hq_token")