[SCRUM-622] refactor(auth): use Powertools secret retrieval for tokens from secrets manager

src/auth_service/refresh_service_accounts_token/lambda_function.py

@@ -3,6 +3,8 @@
 import os
 
 import boto3
+from aws_lambda_powertools.utilities.parameters import get_secret
+from aws_lambda_powertools.utilities.parameters.exceptions import GetParameterError
 from botocore.exceptions import ClientError
 
 # Initialize logger
@@ -59,10 +61,11 @@ def refresh_service_account_access_token(service_account: str) -> dict:
         logger.info("Starting token refresh for service account: %s", service_account)
 
         # Retrieve service account password from Secrets Manager
-        password_response = secrets_client.get_secret_value(
-            SecretId=get_secret_path(service_account, "password")
+        password_secret = get_secret(
+            get_secret_path(service_account, "password"),
+            transform="json",
         )
-        password = json.loads(password_response["SecretString"])["password"]
+        password = password_secret["password"]
 
         # Prepare login payload with complete API Gateway format
         login_payload = {
@@ -122,7 +125,7 @@ def refresh_service_account_access_token(service_account: str) -> dict:
             "message": "Token refresh successful",
         }
 
-    except (ClientError, ValueError) as e:
+    except (ClientError, GetParameterError, ValueError) as e:
         logger.error(
             "Token refresh failed for service account %s: %s", service_account, str(e)
         )

src/auth_service/refresh_service_accounts_token/requirements.txt

@@ -0,0 +1 @@
+aws-lambda-powertools==3.30.0

src/rsvp_service/get_rsvp_status/jwt_util.py

@@ -1,17 +1,13 @@
 import os
 
-import boto3
 import jwt
+from aws_lambda_powertools.utilities.parameters import get_secret
 
 
 class AuthenticationError(Exception):
     pass
 
 
-_secretsmanager_client = boto3.client("secretsmanager")
-_jwt_secret_cache = None
-
-
 def _extract_token(headers):
     if not headers:
         raise AuthenticationError("Missing Authorization header")
@@ -44,19 +40,12 @@ def decode_rsvp_token(headers):
 
 
 def _get_jwt_secret():
-    global _jwt_secret_cache
-
-    if _jwt_secret_cache is not None:
-        return _jwt_secret_cache
-
     jwt_secret_arn = os.getenv("JWT_SECRET_ARN")
     if not jwt_secret_arn:
         raise RuntimeError("JWT_SECRET_ARN environment variable is not set")
 
-    response = _secretsmanager_client.get_secret_value(SecretId=jwt_secret_arn)
-    secret_string = response.get("SecretString")
-    if not secret_string:
+    secret = get_secret(jwt_secret_arn)
+    if not secret:
         raise RuntimeError("JWT secret is empty")
 
-    _jwt_secret_cache = secret_string
-    return _jwt_secret_cache
+    return secret

src/rsvp_service/get_rsvp_status/requirements.txt

@@ -1 +1,2 @@
 pyjwt==2.8.0
+aws-lambda-powertools==3.30.0

src/rsvp_service/update_rsvp/jwt_util.py

@@ -1,17 +1,13 @@
 import os
 
-import boto3
 import jwt
+from aws_lambda_powertools.utilities.parameters import get_secret
 
 
 class AuthenticationError(Exception):
     pass
 
 
-_secretsmanager_client = boto3.client("secretsmanager")
-_jwt_secret_cache = None
-
-
 def _extract_token(headers):
     if not headers:
         raise AuthenticationError("Missing Authorization header")
@@ -44,19 +40,12 @@ def decode_rsvp_token(headers):
 
 
 def _get_jwt_secret():
-    global _jwt_secret_cache
-
-    if _jwt_secret_cache is not None:
-        return _jwt_secret_cache
-
     jwt_secret_arn = os.getenv("JWT_SECRET_ARN")
     if not jwt_secret_arn:
         raise RuntimeError("JWT_SECRET_ARN environment variable is not set")
 
-    response = _secretsmanager_client.get_secret_value(SecretId=jwt_secret_arn)
-    secret_string = response.get("SecretString")
-    if not secret_string:
+    secret = get_secret(jwt_secret_arn)
+    if not secret:
         raise RuntimeError("JWT secret is empty")
 
-    _jwt_secret_cache = secret_string
-    return _jwt_secret_cache
+    return secret

src/rsvp_service/update_rsvp/requirements.txt

@@ -1 +1,2 @@
 pyjwt==2.8.0
+aws-lambda-powertools==3.30.0

src/webhook_service/trigger_webhook/email_service.py

@@ -4,7 +4,7 @@
 
 import requests
 from config import Config
-from utils import SecretsManager
+from utils import get_access_token
 
 # Initialize logger
 logger = logging.getLogger(__name__)
@@ -17,9 +17,6 @@
 class EmailService:
     """Class to handle email service operations"""
 
-    def __init__(self):
-        self.secrets_manager = SecretsManager()
-
     def prepare_email_body(
         self, webhook_details: dict[str, Any], recipient_email: str
     ) -> dict[str, Any]:
@@ -49,7 +46,7 @@ def prepare_email_body(
     def send_email(self, email_body: dict[str, Any]) -> dict[str, Any]:
         """Send an email using the send email API"""
         try:
-            access_token = self.secrets_manager.get_access_token("surveycake")
+            access_token = get_access_token("surveycake")
             logger.info("Send email API endpoint: %s", Config.SEND_EMAIL_API_ENDPOINT)
 
             response = requests.post(

src/webhook_service/trigger_webhook/requirements.txt

@@ -1,2 +1,3 @@
 requests
 pycryptodome
+aws-lambda-powertools==3.30.0

src/webhook_service/trigger_webhook/utils.py

@@ -5,7 +5,7 @@
 import json
 from decimal import Decimal
 
-import boto3
+from aws_lambda_powertools.utilities.parameters import get_secret
 from config import Config
 from Crypto.Cipher import AES
 
@@ -19,26 +19,14 @@ def default(self, o):
         return super().default(o)
 
 
-class SecretsManager:
-    """Class to handle the Secrets Manager operations"""
-
-    def __init__(self):
-        self.client = boto3.client("secretsmanager")
-
-    def get_secret_path(self, service_account: str, secret_type: str) -> str:
-        """Get the secret path based on the service account and secret type"""
-        return f"aws-educate-tpet/{Config.ENVIRONMENT}/service-accounts/{service_account}/{secret_type}"
-
-    def get_access_token(self, service_account: str) -> str:
-        """Get the access token from the Secrets Manager"""
-        try:
-            response = self.client.get_secret_value(
-                SecretId=self.get_secret_path(service_account, "access-token")
-            )
-            return json.loads(response["SecretString"])["access_token"]
-        except Exception as e:
-            print(f"Failed to retrieve access token: {str(e)}")
-            raise
+def get_access_token(service_account: str) -> str:
+    # force_fetch=True: access tokens are rotated by auth_service; never serve a stale cached value
+    secret = get_secret(
+        f"aws-educate-tpet/{Config.ENVIRONMENT}/service-accounts/{service_account}/access-token",
+        transform="json",
+        force_fetch=True,
+    )
+    return secret["access_token"]
 
 
 class CryptoHandler: