Автоматизация управления разрешениями AWS Identity Center

Description: AWS SSO Automation Components

Parameters:
  ManagedResourcePrefix:
    Type: String
  InstanceArn:
    Type: String
  SMTPNotifyAddress:
    Type: String
  TopicName:
    Type: String
    Default: "AssigmentTopic"

Resources:
  NewSSOGroupEventRule:
    Type: AWS::Events::Rule
    Properties:
      Description: Trigger for when a new SSO Group is propagated from external IdP via SCIM
      EventPattern:
        source:
        - aws.sso-directory
        detail-type:
        - AWS API Call via CloudTrail
        detail:
          eventSource:
          - sso-directory.amazonaws.com
          eventName:
          - CreateGroup
      Targets:
        - Arn: !GetAtt SsoAssignGroupsFunction.Arn
          Id: sso-assign-group-function
  ExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
        - Effect: Allow
          Principal:
            Service: lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
      - PolicyName: SSOandOrgPermissions
        PolicyDocument:
          Version: 2012-10-17
          Statement:
          - Effect: Allow
            Action:
            - sso-directory:Describe*
            - sso-directory:Get*
            - sso-directory:List*
            - sso-directory:Describe*
            - sso-directory:Search*
            - sso:Describe*
            - sso:Get*
            - sso:List*
            - sso:CreateAccountAssignment
            - sso:ProvisionPermissionSet
            - identitystore:List*
            - identitystore:Describe*
            - organizations:ListAccounts
            - sns:Publish
            Resource: '*'
  AssignmentTopic:
    Type: AWS::SNS::Topic
    Properties:
      Subscription:
        - Endpoint: !Ref SMTPNotifyAddress
          Protocol: "email"
      TopicName: !Ref TopicName
  SsoAssignGroupsFunction:
    Type: AWS::Lambda::Function
    Properties:
      Code:
        ZipFile: |
          """ This function is intended to be a standalone Lambda function for SSO Permission Set
          to AWS Account Assignment """

          import re
          import os
          import logging
          import json
          from time import sleep
          import boto3
          import traceback

          logger = logging.getLogger()
          logger.setLevel(logging.DEBUG)


          # This function assumes that groups are named like aws_<account-name>_<a|r|p>
          # see the documentation of this function for details.
          CONST_PREFIX = "aws"
          g_account_pattern = rf"^{CONST_PREFIX}_(S*)_(S*)"

          PSET_NAME_MAPPING_DICT = {
              "r": "AWSReadOnlyAccess",
              "a": "AWSAdministratorAccess",
              "p": "AWSPowerUserAccess",
          }

          sso_admin_client = boto3.client("sso-admin")
          org_client = boto3.client("organizations")
          sns_resource = boto3.resource("sns")


          def list_permission_sets(sso_instance_arn) -> dict:
              """Returns a dictionary of permissionssets for a given SSO Instance."""
              perm_set_dict = {}
              response = sso_admin_client.list_permission_sets(InstanceArn=sso_instance_arn)
              results = response["PermissionSets"]
              while "NextToken" in response:
                  response = sso_admin_client.list_permission_sets(
                      InstanceArn=sso_instance_arn, NextToken=response["NextToken"]
                  )
                  results.extend(response["PermissionSets"])

              for permission_set in results:
                  perm_description = sso_admin_client.describe_permission_set(
                      InstanceArn=sso_instance_arn, PermissionSetArn=permission_set
                  )
                  perm_set_dict[perm_description["PermissionSet"]["Name"]] = permission_set
              return perm_set_dict


          def list_aws_accounts() -> list:
              """Returns a list of account dictionaries containing  name id of each account"""
              account_list = []
              paginator = org_client.get_paginator("list_accounts")
              page_iterator = paginator.paginate()

              for page in page_iterator:
                  for acct in page["Accounts"]:
                      # only add active accounts
                      if acct["Status"] == "ACTIVE":
                          data = {"name": acct["Name"], "id": acct["Id"]}
                          account_list.append(data)
              #logger.debug("List of accounts: %s", account_list)
              return account_list


          def lambda_handler(event, context):
              """Main method for Lambda function, will handle the IAM Identity Center permission set to IAM Identity Center directiry group and AWS account mapping"""
              logger.debug("Invoked with event: %s", event)
              try:
                  group_display_name = event["detail"]["responseElements"]["group"]["displayName"]
                  if group_display_name == "":
                      logger.debug(
                          "Recieved SCIM CreateGroup event for roup name '%s'", group_display_name
                      )
                      raise Exception("Event did not contain the group display name property")

                  result = re.search(g_account_pattern, group_display_name)
                  print(result)
                  if not result:
                      logger.error(
                          "Security group: '%s' does not matching naming convention for account assignment. REGEX retourned matches: %s",
                          group_display_name, result,
                      )
                      raise Exception(
                          "Security group does not match convention for account assignment automation"
                      )

                  account_name = result.group(1)
                  short_name_pset = result.group(2)  # "a" "p" or "r"

                  if short_name_pset not in PSET_NAME_MAPPING_DICT:
                      logger.error(
                          "Short name '%s' for permission set is not known", short_name_pset
                      )
                      raise Exception("Short name for permission set is not known")

                  permission_set_name = PSET_NAME_MAPPING_DICT[short_name_pset]
                  logger.debug(
                      "Searching for account '%s' and permission set '%s'",
                      account_name,
                      permission_set_name,
                  )

                  accounts = list_aws_accounts()

                  instance_arn = os.getenv("INSTANCE_ARN")
                  logger.info("IAM Identity Center Instance ARN is configured '%s'", instance_arn)

                  if instance_arn is None:
                      raise Exception("No IAM Idenity Center Instance ARN is configured.")

                  logger.debug("Found IAM Idenity Center instance arn '%s'", instance_arn)
                  permission_sets = list_permission_sets(instance_arn)

                  logger.debug("Permission sets found '%s'", permission_sets)

                  account_id, permission_set_arn, account_name = None, None, None
                  for account in accounts:
                      logger.info("Id of desired account is '%s' ", account["id"])
                      account_id = account.get("id")
                      account_name = account.get("name")

                  if account_id is None:
                      logger.error("Can't find desired account '%s'", account_name)
                      raise Exception("Clould not find account")

                  for name, arn in permission_sets.items():
                      if name == permission_set_name:
                          logger.info("ARN of desired permission set is '%s'", arn)
                          permission_set_arn = arn

                  if permission_set_arn is None:
                      logger.error("Can't find desired permission set %s", permission_set_arn)
                      raise Exception("Can't find desired permission set")

                  principal_id = event["detail"]["responseElements"]["group"]["groupId"]
                  logger.info("PrincipalId of the group is: %s", principal_id)
                  if principal_id is None:
                      logger.error("Could not retrieve the princiapal id of group %s", principal_id)
                      raise Exception("Could not retrieve the princial id of the group")

                  request = {
                      "InstanceArn": instance_arn,
                      "TargetId": account_id,
                      "TargetType": "AWS_ACCOUNT",
                      "PermissionSetArn": permission_set_arn,
                      "PrincipalType": "GROUP",
                      "PrincipalId": principal_id,  #AWS IAM Identity Center group identifier
                  }

                  cracct_response = sso_admin_client.create_account_assignment(**request)
                  cracct_request_id = cracct_response["AccountAssignmentCreationStatus"][
                      "RequestId"
                  ]

                  for tries in range(5):
                      # The docs explain the following valid status states "IN_PROGRESS"|"FAILED"|"SUCCEEDED"
                      ps_prov_set_status = (
                          sso_admin_client.describe_account_assignment_creation_status(
                              InstanceArn=instance_arn,
                              AccountAssignmentCreationRequestId=cracct_request_id,
                          )
                      )
                      logger.info("Assignment attempt %s", tries)
                      status = ps_prov_set_status["AccountAssignmentCreationStatus"]["Status"]

                      if status == "IN_PROGRESS":
                          logger.info("Assignment is in progress")
                          logger.info("Sleeping for 5 seconds")
                          sleep(5.0)
                          continue
                      if status == "FAILED":
                          logger.error("Account assigned has failed")
                          raise Exception("Account assignment has failed")
                          return

                  logger.info(
                      "SUCCESS: Security Group: %s assigned to Account: %s with permission set: %s",
                      group_display_name,
                      account_name,
                      permission_set_name,
                  )
              except Exception as err:
                  message = {"Exception Details": str(err),
                                    "event": event}
                  return message
                  sns_topic = os.getenv("SNS_TOPIC")
                  topic = sns_resource.Topic(sns_topic)
                  topic.publish(
                      Message=json.dumps(message),
                      Subject="Account Association Operation Failed: SSO Automation",
                  )
                  logger.info("Error notification sent via SNS")

      Handler: 'assign_group_to_account.lambda_handler'
      Role: !GetAtt ExecutionRole.Arn
      Runtime: 'python3.9'
      MemorySize: 128
      Timeout: 900
      Environment:
        Variables:
          SNS_TOPIC: !Ref AssignmentTopic
          INSTANCE_ARN: !Ref InstanceArn
  EventsFunctionPermission:
    Type: AWS::Lambda::Permission
    Properties:
      Action: lambda:InvokeFunction
      FunctionName: !GetAtt SsoAssignGroupsFunction.Arn
      Principal: events.amazonaws.com
      SourceArn: !GetAtt NewSSOGroupEventRule.Arn

`

# References
See https://docs.aws.amazon.com/singlesignon/latest/userguide/provision-automatically.html for details on how to set a external Identity Provider configuration for provisioning.