[plugins/aws][feat] Add support for Cloudformation Stack Sets (#682)

* Cloudformation Stack Set support Part 1/2

plugins/aws/resoto_plugin_aws/accountcollector.py

@@ -4,6 +4,7 @@
 import urllib3
 import json
 import re
+from datetime import datetime, timezone, timedelta
 from functools import lru_cache
 from threading import Lock
 from collections.abc import Mapping
@@ -220,6 +221,10 @@
     "resoto_plugin_aws_collect_cloudformation_stacks_seconds",
     "Time it took the collect_cloudformation_stacks() method",
 )
+metrics_collect_cloudformation_stack_sets = Summary(
+    "resoto_plugin_aws_collect_cloudformation_stack_sets_seconds",
+    "Time it took the collect_cloudformation_stack_sets() method",
+)
 metrics_collect_eks_clusters = Summary(
     "resoto_plugin_aws_collect_eks_clusters_seconds",
     "Time it took the collect_eks_clusters() method",
@@ -290,6 +295,7 @@ def __init__(self, regions: List, account: AWSAccount) -> None:
             "network_interfaces": self.collect_network_interfaces,
             "nat_gateways": self.collect_nat_gateways,
             "rds_instances": self.collect_rds_instances,
+            "cloudformation_stack_sets": self.collect_cloudformation_stack_sets,
             "cloudformation_stacks": self.collect_cloudformation_stacks,
             "eks_clusters": self.collect_eks_clusters,
             "vpc_peering_connections": self.collect_vpc_peering_connections,
@@ -1354,7 +1360,7 @@ def collect_instances(self, region: AWSRegion, graph: Graph) -> None:
                         )
                     )
                     i.add_deferred_connection(
-                        "arn", instance.iam_instance_profile["Arn"]
+                        {"arn": instance.iam_instance_profile["Arn"]}
                     )
                 log.debug(
                     f"Found instance {i.id} of type {i.instance_type} status {i.instance_status}"
@@ -1848,10 +1854,10 @@ def collect_albs(self, region: AWSRegion, graph: Graph) -> None:
                                 f"Queuing deferred connection from Server Certificate {certificate_arn} to ALB {a.id}"
                             )
                             a.add_deferred_connection(
-                                "arn", certificate_arn, parent=False
+                                {"arn": certificate_arn}, parent=False
                             )
                             a.add_deferred_connection(
-                                "arn", certificate_arn, edge_type=EdgeType.delete
+                                {"arn": certificate_arn}, edge_type=EdgeType.delete
                             )
 
             except botocore.exceptions.ClientError:
@@ -1928,10 +1934,10 @@ def collect_elbs(self, region: AWSRegion, graph: Graph) -> None:
                             f"Queuing deferred connection from Server Certificate {ssl_certificate_id} to ELB {e.id}"
                         )
                         e.add_deferred_connection(
-                            "arn", ssl_certificate_id, parent=False
+                            {"arn": ssl_certificate_id}, parent=False
                         )
                         e.add_deferred_connection(
-                            "arn", ssl_certificate_id, edge_type=EdgeType.delete
+                            {"arn": ssl_certificate_id}, edge_type=EdgeType.delete
                         )
             except botocore.exceptions.ClientError:
                 log.exception(f"Some boto3 call failed on resource {elb} - skipping")
@@ -2186,6 +2192,80 @@ def collect_cloudformation_stacks(self, region: AWSRegion, graph: Graph) -> None
             log.debug(f"Found Cloudformation Stack {s.name} ({s.id})")
             graph.add_resource(region, s)
 
+    @metrics_collect_cloudformation_stack_sets.time()
+    def collect_cloudformation_stack_sets(
+        self, region: AWSRegion, graph: Graph
+    ) -> None:
+        log.info(
+            f"Collecting AWS Cloudformation Stack Sets in account {self.account.dname} region {region.id}"
+        )
+
+        session = aws_session(self.account.id, self.account.role)
+        client = session.client("cloudformation", region_name=region.id)
+
+        response = client.list_stack_sets(Status="ACTIVE")
+        stack_sets = response.get("Summaries", [])
+        while response.get("NextToken") is not None:
+            response = client.list_stack_sets(
+                Status="ACTIVE", NextToken=response["NextToken"]
+            )
+            stack_sets.extend(response.get("Summaries", []))
+
+        for stack_set_summary in stack_sets:
+            stack_set = client.describe_stack_set(
+                StackSetName=stack_set_summary["StackSetName"]
+            )
+            stack_set = stack_set.get("StackSet", {})
+            s = AWSCloudFormationStackSet(
+                stack_set["StackSetId"],
+                self.tags_as_dict(stack_set.get("Tags", [])),
+                name=stack_set.get("StackSetName"),
+                _account=self.account,
+                _region=region,
+                stack_set_status=stack_set.get("Status"),
+                stack_set_parameters=self.parameters_as_dict(
+                    stack_set.get("Parameters", [])
+                ),
+                stack_set_capabilities=stack_set.get("Capabilities", []),
+                arn=stack_set.get("StackSetARN"),
+                stack_set_administration_role_arn=stack_set.get(
+                    "AdministrationRoleARN"
+                ),
+                stack_set_execution_role_name=stack_set.get("ExecutionRoleName"),
+                stack_set_drift_detection_details=stack_set.get(
+                    "StackSetDriftDetectionDetails", {}
+                ),
+                stack_set_last_drift_check_timestamp=stack_set.get(
+                    "StackSetDriftDetectionDetails", {}
+                ).get("LastDriftCheckTimestamp"),
+                stack_set_auto_deployment=stack_set.get("AutoDeployment", {}),
+                stack_set_permission_model=stack_set.get("PermissionModel"),
+                stack_set_organizational_unit_ids=stack_set.get(
+                    "OrganizationalUnitIds", []
+                ),
+                stack_set_managed_execution_active=stack_set.get(
+                    "ManagedExecution", {}
+                ).get("Active"),
+            )
+            log.debug(f"Found Cloudformation Stack Set {s.name} ({s.id})")
+            graph.add_resource(region, s)
+            # The following requires a feature in the core that would
+            # allow to add edges between resources of different accounts.
+            # https://github.com/someengineering/resoto/issues/693
+            #
+            # response = client.list_stack_instances(StackName=s.name)
+            # stack_instances = response.get("Summaries", [])
+            # while response.get("NextToken") is not None:
+            #     response = client.list_stack_instances(
+            #         StackName=s.name, NextToken=response["NextToken"]
+            #     )
+            #     stack_instances.extend(response.get("Summaries", []))
+            # for stack_instance in stack_instances:
+            #     stack_instance_region = stack_instance.get("Region")
+            #     stack_instance_account = stack_instance.get("Account")
+            #     stack_instance_stack_id = stack_instance.get("StackId")
+            # create a deferred connection that's being resolved core side
+
     @metrics_collect_eks_clusters.time()
     def collect_eks_clusters(self, region: AWSRegion, graph: Graph) -> None:
         log.info(
@@ -2219,9 +2299,9 @@ def collect_eks_clusters(self, region: AWSRegion, graph: Graph) -> None:
                 log.debug(
                     f"Queuing deferred connection from role {cluster['roleArn']} to {c.kind} {c.id}"
                 )
-                c.add_deferred_connection("arn", cluster["roleArn"], parent=False)
+                c.add_deferred_connection({"arn": cluster["roleArn"]}, parent=False)
                 c.add_deferred_connection(
-                    "arn", cluster["roleArn"], edge_type=EdgeType.delete
+                    {"arn": cluster["roleArn"]}, edge_type=EdgeType.delete
                 )
             graph.add_resource(region, c)
             self.get_eks_nodegroups(region, graph, c)

plugins/aws/resoto_plugin_aws/resources.py

@@ -6,7 +6,7 @@
 from resotolib.graph import Graph
 from resotolib.utils import make_valid_timestamp
 from .utils import aws_client, aws_resource
-from typing import ClassVar
+from typing import ClassVar, Any
 from dataclasses import dataclass
 from resotolib.logging import log
 
@@ -1135,3 +1135,75 @@ def delete_tag(self, key) -> bool:
         client = aws_client(self, "cloudwatch")
         client.untag_resource(ResourceARN=self.arn, TagKeys=[key])
         return True
+
+
+@dataclass(eq=False)
+class AWSCloudFormationStackSet(AWSResource, BaseResource):
+    kind: ClassVar[str] = "aws_cloudformation_stack_set"
+    description: Optional[str] = None
+    stack_set_status: Optional[str] = None
+    stack_set_parameters: Dict = field(default_factory=dict)
+    stack_set_capabilities: Optional[List[str]] = field(default_factory=list)
+    stack_set_administration_role_arn: Optional[str] = None
+    stack_set_execution_role_name: Optional[str] = None
+    stack_set_drift_detection_details: Optional[Dict[str, Any]] = field(
+        default_factory=dict
+    )
+    stack_set_last_drift_check_timestamp: Optional[datetime] = None
+    stack_set_auto_deployment: Optional[Dict[str, bool]] = field(default_factory=dict)
+    stack_set_permission_model: Optional[str] = None
+    stack_set_organizational_unit_ids: Optional[List[str]] = field(default_factory=list)
+    stack_set_managed_execution_active: Optional[bool] = None
+
+    def delete(self, graph: Graph) -> bool:
+        cf = aws_client(self, "cloudformation", graph)
+        cf.delete_stack_set(StackSetName=self.name)
+        return True
+
+    class ModificationMode(Enum):
+        """Defines Tag modification mode"""
+
+        UPDATE = auto()
+        DELETE = auto()
+
+    def update_tag(self, key, value) -> bool:
+        return self._modify_tag(
+            key, value, mode=AWSCloudFormationStackSet.ModificationMode.UPDATE
+        )
+
+    def delete_tag(self, key) -> bool:
+        return self._modify_tag(
+            key, mode=AWSCloudFormationStackSet.ModificationMode.DELETE
+        )
+
+    def _modify_tag(self, key, value=None, mode=None) -> bool:
+        tags = dict(self.tags)
+        if mode == AWSCloudFormationStackSet.ModificationMode.DELETE:
+            if not self.tags.get(key):
+                raise KeyError(key)
+            del tags[key]
+        elif mode == AWSCloudFormationStackSet.ModificationMode.UPDATE:
+            if self.tags.get(key) == value:
+                return True
+            tags.update({key: value})
+        else:
+            return False
+
+        cf = aws_client(self, "cloudformation")
+        response = cf.update_stack_set(
+            StackSetName=self.name,
+            Capabilities=["CAPABILITY_NAMED_IAM"],
+            UsePreviousTemplate=True,
+            Tags=[{"Key": label, "Value": value} for label, value in tags.items()],
+            Parameters=[
+                {"ParameterKey": parameter, "UsePreviousValue": True}
+                for parameter in self.stack_set_parameters.keys()
+            ],
+        )
+        if response.get("ResponseMetadata", {}).get("HTTPStatusCode", 0) != 200:
+            raise RuntimeError(
+                "Error updating AWS Cloudformation Stack Set"
+                f" {self.dname} for {mode.name} of tag {key}"
+            )
+        self.tags = tags
+        return True

plugins/gcp/resoto_plugin_gcp/collector.py

@@ -679,14 +679,15 @@ def collect_something(
                                         values = value
                                         for value in values:
                                             r.add_deferred_connection(
-                                                attr,
-                                                value,
+                                                {attr: value},
                                                 is_parent,
                                                 edge_type=edge_type,
                                             )
                                     elif isinstance(value, str):
                                         r.add_deferred_connection(
-                                            attr, value, is_parent, edge_type=edge_type
+                                            {attr: value},
+                                            is_parent,
+                                            edge_type=edge_type,
                                         )
                                     else:
                                         log.error(

plugins/k8s/resoto_plugin_k8s/resources/common.py

@@ -90,11 +90,11 @@ def collect(cls, api_client: client.ApiClient, graph: Graph):
                                     values = value
                                     for value in values:
                                         resource.add_deferred_connection(
-                                            attr, value, is_parent
+                                            {attr: value}, is_parent
                                         )
                                 elif isinstance(value, str):
                                     resource.add_deferred_connection(
-                                        attr, value, is_parent
+                                        {attr: value}, is_parent
                                     )
                                 else:
                                     log.error(

resotolib/resotolib/baseresources.py

@@ -493,18 +493,18 @@ def to_json(self):
         return self.__repr__()
 
     def add_deferred_connection(
-        self, attr, value, parent: bool = True, edge_type: EdgeType = EdgeType.default
+        self, search: Dict, parent: bool = True, edge_type: EdgeType = EdgeType.default
     ) -> None:
         self._deferred_connections.append(
-            {"attr": attr, "value": value, "parent": parent, "edge_type": edge_type}
+            {"search": search, "parent": parent, "edge_type": edge_type}
         )
 
     def resolve_deferred_connections(self, graph) -> None:
         if graph is None:
             graph = self._graph
         while self._deferred_connections:
             dc = self._deferred_connections.pop(0)
-            node = graph.search_first(dc["attr"], dc["value"])
+            node = graph.search_first_all(dc["search"])
             edge_type = dc["edge_type"]
             if node:
                 if dc["parent"]: