Implements sso-admin AWS Managed Policies (#7184)

moto/ssoadmin/exceptions.py

@@ -3,9 +3,30 @@
 
 
 class ResourceNotFoundException(JsonRESTError):
-    def __init__(self, message: str = "Account not found") -> None:
+    code = 400
+
+    def __init__(self, message: str = "") -> None:
         super().__init__(
             error_type="ResourceNotFoundException",
             message=message,
-            code="ResourceNotFoundException",
+        )
+
+
+class ConflictException(JsonRESTError):
+    code = 400
+
+    def __init__(self, message: str = "") -> None:
+        super().__init__(
+            error_type="ConflictException",
+            message=message,
+        )
+
+
+class ServiceQuotaExceededException(JsonRESTError):
+    code = 400
+
+    def __init__(self, message: str = "") -> None:
+        super().__init__(
+            error_type="ServiceQuotaExceededException",
+            message=message,
         )

moto/ssoadmin/models.py

@@ -1,13 +1,22 @@
-from typing import Any, Dict, List
+import json
+from typing import Any, Dict, List, Optional
 
 from moto.core import BackendDict, BaseBackend, BaseModel
 from moto.core.utils import unix_time
+from moto.iam.aws_managed_policies import aws_managed_policies_data
 from moto.moto_api._internal import mock_random as random
 from moto.utilities.paginator import paginate
 
-from .exceptions import ResourceNotFoundException
+from .exceptions import (
+    ConflictException,
+    ResourceNotFoundException,
+    ServiceQuotaExceededException,
+)
 from .utils import PAGINATION_MODEL
 
+# https://docs.aws.amazon.com/singlesignon/latest/userguide/limits.html
+MAX_MANAGED_POLICIES_PER_PERMISSION_SET = 20
+
 
 class AccountAssignment(BaseModel):
     def __init__(
@@ -60,6 +69,10 @@ def __init__(
         self.tags = tags
         self.created_date = unix_time()
         self.inline_policy = ""
+        self.managed_policies: List[ManagedPolicy] = list()
+        self.total_managed_policies_attached = (
+            0  # this will also include customer managed policies
+        )
 
     def to_json(self, include_creation_date: bool = False) -> Dict[str, Any]:
         summary: Dict[str, Any] = {
@@ -83,13 +96,25 @@ def generate_id(instance_arn: str) -> str:
         )
 
 
+class ManagedPolicy(BaseModel):
+    def __init__(self, arn: str, name: str):
+        self.arn = arn
+        self.name = name
+
+    def __eq__(self, other: Any) -> bool:
+        if not isinstance(other, ManagedPolicy):
+            return False
+        return self.arn == other.arn
+
+
 class SSOAdminBackend(BaseBackend):
     """Implementation of SSOAdmin APIs."""
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.account_assignments: List[AccountAssignment] = list()
         self.permission_sets: List[PermissionSet] = list()
+        self.aws_managed_policies: Optional[Dict[str, Any]] = None
 
     def create_account_assignment(
         self,
@@ -158,6 +183,26 @@ def _find_account(
                 return account
         raise ResourceNotFoundException
 
+    def _find_managed_policy(self, managed_policy_arn: str) -> ManagedPolicy:
+        """
+        Checks to make sure the managed policy exists.
+        This pulls from moto/iam/aws_managed_policies.py
+        """
+        # Lazy loading of aws managed policies file
+        if self.aws_managed_policies is None:
+            self.aws_managed_policies = json.loads(aws_managed_policies_data)
+
+        policy_name = managed_policy_arn.split("/")[-1]
+        managed_policy = self.aws_managed_policies.get(policy_name, None)
+        if managed_policy is not None:
+            path = managed_policy.get("path", "/")
+            expected_arn = f"arn:aws:iam::aws:policy{path}{policy_name}"
+            if managed_policy_arn == expected_arn:
+                return ManagedPolicy(managed_policy_arn, policy_name)
+        raise ResourceNotFoundException(
+            f"Policy does not exist with ARN: {managed_policy_arn}"
+        )
+
     @paginate(PAGINATION_MODEL)  # type: ignore[misc]
     def list_account_assignments(
         self, instance_arn: str, account_id: str, permission_set_arn: str
@@ -314,5 +359,70 @@ def delete_inline_policy_from_permission_set(
         )
         permission_set.inline_policy = ""
 
+    def attach_managed_policy_to_permission_set(
+        self, instance_arn: str, permission_set_arn: str, managed_policy_arn: str
+    ) -> None:
+        permissionset = self._find_permission_set(
+            instance_arn,
+            permission_set_arn,
+        )
+        managed_policy = self._find_managed_policy(managed_policy_arn)
+
+        permissionset_id = permission_set_arn.split("/")[-1]
+        if managed_policy in permissionset.managed_policies:
+            raise ConflictException(
+                f"Permission set with id {permissionset_id} already has a typed link attachment to a manged policy with {managed_policy_arn}"
+            )
+
+        if (
+            permissionset.total_managed_policies_attached
+            >= MAX_MANAGED_POLICIES_PER_PERMISSION_SET
+        ):
+            permissionset_id = permission_set_arn.split("/")[-1]
+            raise ServiceQuotaExceededException(
+                f"You have exceeded AWS SSO limits. Cannot create ManagedPolicy more than {MAX_MANAGED_POLICIES_PER_PERMISSION_SET} for id {permissionset_id}. Please refer to https://docs.aws.amazon.com/singlesignon/latest/userguide/limits.html"
+            )
+
+        permissionset.managed_policies.append(managed_policy)
+        permissionset.total_managed_policies_attached += 1
+
+    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    def list_managed_policies_in_permission_set(
+        self,
+        instance_arn: str,
+        permission_set_arn: str,
+    ) -> List[ManagedPolicy]:
+        permissionset = self._find_permission_set(
+            instance_arn,
+            permission_set_arn,
+        )
+        return permissionset.managed_policies
+
+    def _detach_managed_policy(
+        self, instance_arn: str, permission_set_arn: str, managed_policy_arn: str
+    ) -> None:
+        # ensure permission_set exists
+        permissionset = self._find_permission_set(
+            instance_arn,
+            permission_set_arn,
+        )
+
+        for managed_policy in permissionset.managed_policies:
+            if managed_policy.arn == managed_policy_arn:
+                permissionset.managed_policies.remove(managed_policy)
+                permissionset.total_managed_policies_attached -= 1
+                return
+
+        raise ResourceNotFoundException(
+            f"Could not find ManagedPolicy with arn {managed_policy_arn}"
+        )
+
+    def detach_managed_policy_from_permission_set(
+        self, instance_arn: str, permission_set_arn: str, managed_policy_arn: str
+    ) -> None:
+        self._detach_managed_policy(
+            instance_arn, permission_set_arn, managed_policy_arn
+        )
+
 
 ssoadmin_backends = BackendDict(SSOAdminBackend, "sso")

moto/ssoadmin/responses.py

@@ -195,3 +195,52 @@ def delete_inline_policy_from_permission_set(self) -> str:
             permission_set_arn=permission_set_arn,
         )
         return json.dumps({})
+
+    def attach_managed_policy_to_permission_set(self) -> str:
+        instance_arn = self._get_param("InstanceArn")
+        permission_set_arn = self._get_param("PermissionSetArn")
+        managed_policy_arn = self._get_param("ManagedPolicyArn")
+        self.ssoadmin_backend.attach_managed_policy_to_permission_set(
+            instance_arn=instance_arn,
+            permission_set_arn=permission_set_arn,
+            managed_policy_arn=managed_policy_arn,
+        )
+        return json.dumps({})
+
+    def list_managed_policies_in_permission_set(self) -> str:
+        instance_arn = self._get_param("InstanceArn")
+        permission_set_arn = self._get_param("PermissionSetArn")
+        max_results = self._get_int_param("MaxResults")
+        next_token = self._get_param("NextToken")
+
+        (
+            managed_policies,
+            next_token,
+        ) = self.ssoadmin_backend.list_managed_policies_in_permission_set(
+            instance_arn=instance_arn,
+            permission_set_arn=permission_set_arn,
+            max_results=max_results,
+            next_token=next_token,
+        )
+
+        managed_policies_response = [
+            {"Arn": managed_policy.arn, "Name": managed_policy.name}
+            for managed_policy in managed_policies
+        ]
+        return json.dumps(
+            {
+                "AttachedManagedPolicies": managed_policies_response,
+                "NextToken": next_token,
+            }
+        )
+
+    def detach_managed_policy_from_permission_set(self) -> str:
+        instance_arn = self._get_param("InstanceArn")
+        permission_set_arn = self._get_param("PermissionSetArn")
+        managed_policy_arn = self._get_param("ManagedPolicyArn")
+        self.ssoadmin_backend.detach_managed_policy_from_permission_set(
+            instance_arn=instance_arn,
+            permission_set_arn=permission_set_arn,
+            managed_policy_arn=managed_policy_arn,
+        )
+        return json.dumps({})

moto/ssoadmin/utils.py

@@ -30,4 +30,11 @@
             "PrincipalType",
         ],
     },
+    "list_managed_policies_in_permission_set": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "result_key": "AttachedManagedPolicies",
+        "unique_attribute": ["arn"],
+    },
 }