feat: Allow to use AWS secretsmanager with managed external secrets

aws/logs_monitoring/README.md

@@ -492,7 +492,7 @@ The Datadog Forwarder is signed by Datadog. To verify the integrity of the Forwa
 : Your [Datadog API key][20], which can be found under **Organization Settings** > **API Keys**. The API Key is stored in AWS Secrets Manager. If you already have a Datadog API Key stored in Secrets Manager, use `DdApiKeySecretArn` instead.
 
 `DdApiKeySecretArn`
-: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. You must store the secret as a plaintext, rather than a key-value pair.
+: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. The secret can be a plaintext string, a JSON object with a `DD_API_KEY` field, or an AWS Secrets Manager managed rotation secret of type `DatadogApiKey` (JSON object with an `apiKey` field).
 
 `DdApiKeySsmParameterName`
 : The name of the SSM parameter containing the Datadog API key. If set, both `DdApiKey` and `DdApiKeySecretArn` are ignored.
@@ -676,7 +676,7 @@ If you are installing the Forwarder manually, convert the parameter names from P
 : Your [Datadog API key][20], which can be found under **Organization Settings** > **API Keys**. The API Key is stored in AWS Secrets Manager. If you already have a Datadog API Key stored in Secrets Manager, use `DD_API_KEY_SECRET_ARN` instead.
 
 `DD_API_KEY_SECRET_ARN`
-: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. You must store the secret as a plaintext, rather than a key-value pair.
+: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. The secret can be a plaintext string, a JSON object with a `DD_API_KEY` field, or an AWS Secrets Manager managed rotation secret of type `DatadogApiKey` (JSON object with an `apiKey` field).
 
 `DD_API_KEY_SSM_NAME`
 : The name of the parameter in AWS Systems Manager (SSM) Parameter Store containing the Datadog API key. Takes precedence over `DD_KMS_API_KEY` and `DD_API_KEY`.

aws/logs_monitoring/settings.py

@@ -180,15 +180,20 @@ def __init__(self, name, pattern, placeholder, enabled=True):
         # Try to parse the secret as JSON
         secret_json = json.loads(secret_string)
 
-        # If it's a JSON object, look for the 'DD_API_KEY' field
+        # If it's a JSON object, look for the 'DD_API_KEY' field (our own
+        # convention), then 'apiKey' (the field name used by AWS Secrets
+        # Manager's managed rotation for the Datadog API key secret type)
         if "DD_API_KEY" in secret_json:
             DD_API_KEY = secret_json["DD_API_KEY"]
             logger.debug(
                 "Successfully retrieved the Datadog API key from 'DD_API_KEY'."
             )
+        elif "apiKey" in secret_json:
+            DD_API_KEY = secret_json["apiKey"]
+            logger.debug("Successfully retrieved the Datadog API key from 'apiKey'.")
         else:
             logger.error(
-                "The secret does not contain the 'DD_API_KEY' field. "
+                "The secret does not contain the 'DD_API_KEY' or 'apiKey' field. "
                 "Please ensure the secret is in the correct format. "
                 "Not setting the Datadog API key."
             )

aws/logs_monitoring/template.yaml

@@ -15,7 +15,7 @@ Parameters:
     Type: String
     AllowedPattern: "arn:.*:secretsmanager:.*"
     Default: "arn:aws:secretsmanager:DEFAULT"
-    Description: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. You must store the secret as a plaintext, rather than a key-value pair.
+    Description: The ARN of the secret storing the Datadog API key, if you already have it stored in Secrets Manager. The secret can be a plaintext string, a JSON object with a DD_API_KEY field, or an AWS Secrets Manager managed rotation secret of type DatadogApiKey (JSON object with an apiKey field).
   DdApiKeySsmParameterName:
     Type: String
     Default: "/my/parameter/path"

aws/logs_monitoring/tests/test_settings.py

@@ -1,4 +1,8 @@
+import json
+import os
+import sys
 import unittest
+from importlib import reload
 from unittest.mock import MagicMock, patch
 
 from settings import is_api_key_valid
@@ -11,6 +15,50 @@ class _FakeNetworkError(Exception):
     pass
 
 
+SECRET_ARN = "arn:aws:secretsmanager:us-east-1:123456789012:secret:test"
+
+# A plaintext secret that happens to be all-digits (like VALID_API_KEY) is
+# valid JSON too (it parses as an int), which exercises a different, unrelated
+# code path. Use a hex-looking key with a letter so json.loads() reliably
+# raises JSONDecodeError and the plaintext fallback branch is what's tested.
+PLAINTEXT_API_KEY = "abcd1234abcd1234abcd1234abcd1234"
+
+
+class TestApiKeySecretArn(unittest.TestCase):
+    def _reload_with_secret_string(self, secret_string):
+        mock_secretsmanager = MagicMock()
+        mock_secretsmanager.get_secret_value.return_value = {
+            "SecretString": secret_string
+        }
+        with patch("boto3.client", return_value=mock_secretsmanager):
+            reload(sys.modules["settings"])
+        return sys.modules["settings"].DD_API_KEY
+
+    @patch.dict(os.environ, {"DD_API_KEY_SECRET_ARN": SECRET_ARN})
+    def test_plaintext_secret(self):
+        self.assertEqual(
+            self._reload_with_secret_string(PLAINTEXT_API_KEY), PLAINTEXT_API_KEY
+        )
+
+    @patch.dict(os.environ, {"DD_API_KEY_SECRET_ARN": SECRET_ARN})
+    def test_dd_api_key_json_field(self):
+        secret_string = json.dumps({"DD_API_KEY": VALID_API_KEY})
+        self.assertEqual(self._reload_with_secret_string(secret_string), VALID_API_KEY)
+
+    @patch.dict(os.environ, {"DD_API_KEY_SECRET_ARN": SECRET_ARN})
+    def test_aws_managed_secret_api_key_field(self):
+        # AWS Secrets Manager's managed rotation for the Datadog API key
+        # secret type stores the key under 'apiKey', alongside 'apiKeyId'.
+        secret_string = json.dumps({"apiKey": VALID_API_KEY, "apiKeyId": "some-uuid"})
+        self.assertEqual(self._reload_with_secret_string(secret_string), VALID_API_KEY)
+
+    @patch.dict(os.environ, {"DD_API_KEY_SECRET_ARN": SECRET_ARN})
+    def test_dd_api_key_field_takes_precedence_over_api_key(self):
+        other_key = "2" * 32
+        secret_string = json.dumps({"DD_API_KEY": VALID_API_KEY, "apiKey": other_key})
+        self.assertEqual(self._reload_with_secret_string(secret_string), VALID_API_KEY)
+
+
 class TestIsApiKeyValid(unittest.TestCase):
     @patch("settings.DD_API_KEY", VALID_API_KEY)
     @patch("settings.requests.Session")