chore: Service Configuration Manager role based access improvements (#…

…1919)

twilio-iac/scripts/python_tools/src/service_configuration/config.py

@@ -4,7 +4,7 @@
 from typing import NotRequired, TypedDict, Unpack
 from ..aws import SSMClient
 from ..twilio import Twilio
-from .constants import AWS_ROLE_ARN
+from .constants import AWS_ROLE_ARNS, get_aws_role_arn
 from .service_configuration import ServiceConfiguration
 from .remote_syncer import RemoteSyncer
 
@@ -149,6 +149,7 @@ class ConfigDict(TypedDict):
     skip_lock: bool
     sync_action: bool
     argument: str
+    aws_role_arn: str
 
 
 class InitServiceConfigurationArgsDict(TypedDict):
@@ -174,6 +175,7 @@ def __init__(self):
             'json_available': False,
             'sync_action': False,
             'syncers': [],
+            'aws_role_arn': None
         }
 
         self.init_arg_parser()
@@ -188,8 +190,9 @@ def __getattr__(self, name: str):
         except KeyError:
             raise AttributeError(f'No such attribute: {name}')
 
-    def get_ssm_client(self):
-        return SSMClient(AWS_ROLE_ARN)
+    def get_ssm_client(self, environment: str):
+        aws_role_arn = get_aws_role_arn(environment)
+        return SSMClient(aws_role_arn)
 
     def init_arg_parser(self) -> None:
         self._arg_parser = ArgumentParser()
@@ -248,7 +251,8 @@ def add_helpline(
         self._config['helplines'][helpline_code_lower][environment] = service_config
 
     def init_service_config(self, **kwargs: Unpack[InitServiceConfigurationArgsDict]):
-        ssm_client = None if self.auth_token else self.get_ssm_client()
+        environment = kwargs['environment']
+        ssm_client = None if self.auth_token else self.get_ssm_client(environment)
         twilio_client = Twilio(ssm_client=ssm_client,  **kwargs)
         service_config = ServiceConfiguration(
             twilio_client=twilio_client,
@@ -279,7 +283,7 @@ def init_service_configs_for_helpline(
                 'Could not find helpline code or environment. Please provide helpline code and environment')
 
         if not account_sid:
-            account_sid, _ = self.get_ssm_client().get_twilio_creds_from_ssm(
+            account_sid, _ = self.get_ssm_client(environment).get_twilio_creds_from_ssm(
                 environment, helpline_code)
 
         self.init_service_config(
@@ -305,7 +309,7 @@ def init_syncers(self):
 
     def init_service_configs_for_environment(self, environment: str):
         print(f'Initializing service configurations for {environment}')
-        for hl in self.get_ssm_client().get_helplines_for_env(environment):
+        for hl in self.get_ssm_client(environment).get_helplines_for_env(environment):
             # if helpline_code is set, only initialize service configs for that helpline across all environments
             if (self.helpline_code and hl['helpline_code'].lower() != self.helpline_code.lower()):
                 continue

twilio-iac/scripts/python_tools/src/service_configuration/constants.py

@@ -1,2 +1,10 @@
 """ Constants for the service configuration manager """
-AWS_ROLE_ARN: str = 'arn:aws:iam::712893914485:role/twilio-iac-service-config-manager'
+AWS_ROLE_ARNS: dict[str, str] = {
+    'developer': 'arn:aws:iam::712893914485:role/twilio-iac-service-config-developer',
+    'manager': 'arn:aws:iam::712893914485:role/twilio-iac-service-config-manager',
+}
+
+def get_aws_role_arn(environment: str):
+    role_key = "manager" if environment == "production" else "developer"
+    aws_role_arn = AWS_ROLE_ARNS.get(role_key)
+    return aws_role_arn

twilio-iac/scripts/python_tools/src/service_configuration/service_configuration.py

@@ -6,7 +6,7 @@
 from typing import TypedDict, Unpack
 from ..aws import SSMClient
 from ..twilio import Twilio
-from .constants import AWS_ROLE_ARN
+from .constants import AWS_ROLE_ARNS, get_aws_role_arn
 from .version import Version
 
 JSON_PATH_ROOT = "/app/twilio-iac/helplines"
@@ -142,7 +142,7 @@ class LocalConfigsDict(TypedDict):
 
 
 class ServiceConfiguration():
-
+    
     def __init__(self, **kwargs: Unpack[InitArgsDict]) -> None:
         self.local_state: dict[str, object] = {}
         self.new_state: dict[str, object] = {}
@@ -156,6 +156,7 @@ def __init__(self, **kwargs: Unpack[InitArgsDict]) -> None:
         self.account_sid = self._twilio_client.account_sid
         self.helpline_code = self._twilio_client.helpline_code
         self.environment = self._twilio_client.environment
+        self.aws_role_arn = get_aws_role_arn(self.environment)
         self.remote_state: dict[str,
                                 object] = self._twilio_client.get_flex_configuration()
         self.init_version()
@@ -165,7 +166,7 @@ def __init__(self, **kwargs: Unpack[InitArgsDict]) -> None:
         self.init_plan()
 
     def get_ssm_client(self):
-        return SSMClient(AWS_ROLE_ARN)
+        return SSMClient(self.aws_role_arn)
 
     def init_region(self):
         try:
@@ -270,6 +271,7 @@ def init_version(self):
             helpline_code=self.helpline_code,
             state=self.remote_state,
             skip_lock=self.skip_lock,
+            aws_role_arn=self.aws_role_arn
         )
 
     def get_config_path(self, type: str) -> str:
@@ -294,6 +296,7 @@ def apply(self):
             helpline_code=self.helpline_code,
             state=self.new_state,
             skip_lock=True,
+            aws_role_arn=self.aws_role_arn
         )
 
     def cleanup(self):

twilio-iac/scripts/python_tools/src/service_configuration/version.py

@@ -4,7 +4,6 @@
 from datetime import datetime
 from typing import NotRequired, TypedDict, Unpack
 from ..aws import S3Client
-from .constants import AWS_ROLE_ARN
 
 """
 We keep track of every version of the service configuration in S3 so we have a change log.
@@ -25,6 +24,7 @@ class InitArgsDict(TypedDict):
     helpline_code: str
     state: dict[str, str]
     skip_lock: NotRequired[bool | None]
+    aws_role_arn: str
 
 
 class Version():
@@ -33,7 +33,7 @@ def __init__(self, **kwargs: Unpack[InitArgsDict]):
         self.helpline_code = kwargs['helpline_code']
         self.state = kwargs['state']
         self.skip_lock = kwargs.get('skip_lock') or False
-        self.s3_client = S3Client(AWS_ROLE_ARN)
+        self.s3_client = S3Client(kwargs.get('aws_role_arn'))
         self.init_ip()
         self.init_sha()
         self.init_s3_paths()