AWS firewall request automation Design and implementation

Rules and routes decide

What this code does: If any of the security group checks evaluated earlier fails, the verdict becomes a failure with “Security group rule needed” as the reason. When the source is an AWS resource but the destination is outside AWS, it finds the route table of the source subnet and checks for a route to the destination, recording the next hop if one exists or which table lacks it if not. A missing route is recorded separately as “Route needed”, which separates cases that opening a security group would never fix.

        if any(not c.ok for c in verdict.checks):
            verdict.ok = False
            verdict.issues.append("Security group rule needed")

        if source.resources and not target.resources:
            for resource in source.resources:
                ec2 = self.sessions.client("ec2", resource.account)
                result = route_analysis.evaluate(
                    route_analysis.find_table(ec2, resource.subnet_id, resource.vpc_id), dst
                )
                detail = (
                    f"via {route_analysis.next_hop(result.routes[0])}"
                    if result.ok
                    else f"no route in {result.table_id or 'any table'}"
                )
                verdict.checks.append(Check("route", resource.label, result.ok, detail))
                if not result.ok:
                    verdict.ok = False
                    verdict.issues.append("Route needed")
        return verdict

Apply only adds rules

What this code does: Calls only the API that adds an inbound or outbound rule, depending on the planned direction. There is no call that deletes or edits a rule, so this tool cannot cut existing traffic. If the same rule already exists and AWS returns InvalidPermission.Duplicate, it is treated as “already present” and skipped, while any other error is raised as is instead of being hidden.

    def apply(self, change: Change) -> str:
        ec2 = self.sessions.client("ec2", change.account)
        call = (
            ec2.authorize_security_group_ingress
            if change.direction == "ingress"
            else ec2.authorize_security_group_egress
        )
        try:
            call(GroupId=change.sg_id, IpPermissions=change.permissions)
        except ClientError as error:
            if error.response["Error"]["Code"] == "InvalidPermission.Duplicate":
                return "already present"
            raise
        return "applied"