feat: add sg to ingress & egress

Author: Komalis

easyecs/cloudformation/template/__init__.py

@@ -128,6 +128,20 @@ def create_load_balancer(stack, ecs_manifest: EcsFileModel, vpc):
                             ),
                             description=egress_rule.name,
                         )
+                    elif egress_rule.security_group_id:
+                        lb_security_group.add_egress_rule(
+                            peer=SecurityGroup.from_security_group_id(
+                                stack,
+                                "egress_rule_sg",
+                                security_group_id=egress_rule.security_group_id,
+                            ),
+                            connection=(
+                                Port.tcp(egress_rule.port)
+                                if egress_rule.port != -1
+                                else Port.all_traffic()
+                            ),
+                            description=egress_rule.name,
+                        )
             if ecs_manifest.load_balancer.security_group_rules.ingress:
                 for (
                     ingress_rule
@@ -156,6 +170,20 @@ def create_load_balancer(stack, ecs_manifest: EcsFileModel, vpc):
                             ),
                             description=ingress_rule.name,
                         )
+                    elif ingress_rule.security_group_id:
+                        lb_security_group.add_ingress_rule(
+                            peer=SecurityGroup.from_security_group_id(
+                                stack,
+                                "ingress_rule_sg",
+                                security_group_id=ingress_rule.security_group_id,
+                            ),
+                            connection=(
+                                Port.tcp(ingress_rule.port)
+                                if ingress_rule.port != -1
+                                else Port.all_traffic()
+                            ),
+                            description=ingress_rule.name,
+                        )
         listener = lb.add_listener(
             "NlbListener", port=ecs_manifest.load_balancer.listener_port
         )

easyecs/cloudformation/template/task_definition.py

@@ -1,5 +1,5 @@
 import re
-from easyecs.model.ecs import EcsFileSecretModelV2
+from easyecs.model.ecs import EcsFileSecretModel, EcsFileSecretModelV2
 
 
 def create_task_definition(
@@ -121,13 +121,13 @@ def extract_container_config(stack, container_definition, log_configuration, run
         command = ["sleep", "infinity"]
 
     environment = {}
-    if isinstance(container_definition.env, List):
+    if isinstance(container_definition.env, list):
         environment = {
             env_definition.name: env_definition.value
             for env_definition in container_definition.env
             if env_definition.active
         }
-    elif isinstance(container_definition.env, Dict):
+    elif isinstance(container_definition.env, dict):
         environment = container_definition.env
 
     secrets = extract_secrets(stack, container_definition.secrets, name)
@@ -173,16 +173,22 @@ def extract_secrets(stack, secret_definitions, container_name):
             ecs_secret = ECSSecret.from_secrets_manager(secret, secret_definition.field)
             secrets[secret_name] = ecs_secret
         elif isinstance(secret_definition, EcsFileSecretModelV2):
-            arn_fields = dict(re.finditer(r"^arn:aws:secretsmanager:(?P<region_name>[a-z0-9-]+):(?P<account_id>\d{12}):secret:(?P<secret_name>[^:]+)(?::(?P<field>[^:]*))?(?::([^:]*))?(?::([^:]*))?$", secret_definition.valueFrom))
+            arn_fields = list(
+                re.finditer(
+                    r"^^(?P<secret_complete_arn>arn:aws:secretsmanager:(?P<region_name>[a-z0-9-]+):(?P<account_id>\d{12}):secret:(?P<secret_name>[^:]+))(?::(?P<field>[^:]*))?(?::([^:]*))?(?::([^:]*))?$",  # noqa
+                    secret_definition.valueFrom,
+                )
+            )
             if not arn_fields:
                 raise ValueError(f"Invalid ARN format: {secret_definition.valueFrom}")
-            field = arn_fields[0].group_dict()["field"]
-            import pdb; pdb.set_trace()
+            secret_complete_arn = arn_fields[0].groupdict()["secret_complete_arn"]
+            field = arn_fields[0].groupdict()["field"]
+            print(secret_definition.valueFrom)
             secret_name = secret_definition.name
             secret = Secret.from_secret_complete_arn(
-                stack, f"{secret_name}_{container_name}", secret_definition.valueFrom
+                stack, f"{secret_name}_{container_name}", secret_complete_arn
             )
-            ecs_secret = ECSSecret.from_secrets_manager(secret)
+            ecs_secret = ECSSecret.from_secrets_manager(secret, field=field)
             secrets[secret_name] = ecs_secret
         else:
             raise Exception("Unsupported secret type")

easyecs/model/ecs.py

@@ -155,13 +155,30 @@ class SecurityGroupRule(BaseModel):
     port: int
     cidr: Optional[str] = None
     prefix_list: Optional[str] = None
+    security_group_id: Optional[str] = None
 
     @model_validator(mode="after")
     def validate_cidr(self):
-        if self.prefix_list is not None and self.cidr is not None:
-            raise ValueError("A rule is either a CIDR or a prefix list, not both!")
-        if self.prefix_list is None and self.cidr is None:
-            raise ValueError("A rule is either a CIDR or a prefix list, not none!")
+        has_at_least_two_true = lambda lst: sum(lst) >= 2  # noqa: E731
+        if has_at_least_two_true(
+            [
+                self.prefix_list is not None,
+                self.cidr is not None,
+                self.security_group_id is not None,
+            ]
+        ):
+            raise ValueError(
+                "A rule is either a CIDR, a security group id or a prefix list!"
+            )
+        if (
+            self.prefix_list is None
+            and self.cidr is None
+            and self.security_group_id is None
+        ):
+            raise ValueError(
+                "A rule is either a CIDR, a security group id or a prefix list, not"
+                " none!"
+            )
         return self