How to use describe_traffic_mirror_filters method in localstack

Best Python code snippet using localstack_python

mirroring.py

Source:mirroring.py

...106        return response["TrafficMirrorFilter"]["TrafficMirrorFilterId"]107    @staticmethod108    def find_traffic_mirror_filter_ids_by_reservation_id(ec2_client, reservation_id):109        traffic_mirror_filters = []110        response = ec2_client.describe_traffic_mirror_filters(111            Filters=[112                {"Name": "tag:ReservationId", "Values": [str(reservation_id)]},113            ],114            MaxResults=123,115        )116        traffic_mirror_filters.extend(response["TrafficMirrorFilters"])117        while "NextToken" in response:118            response = ec2_client.describe_traffic_mirror_filters(119                NextToken=response["NextToken"]120            )121            traffic_mirror_filters.extend(response["TrafficMirrorFilters"])122        return [123            traffic_filter["TrafficMirrorFilterId"]124            for traffic_filter in traffic_mirror_filters125        ]126    # region traffic mirror rules127    def create_filter_rules(self, ec2_client, fulfillment):128        """# noqa129        creates filter rules requested for a particular session130        :param cloudshell.cp.aws.models.traffic_mirror_fulfillment.TrafficMirrorFulfillment fulfillment:131        """132        i = 0133        for rule in fulfillment.filter_rules:134            i += 1135            description = f"filter_rule_{i}_{fulfillment.traffic_mirror_filter_id}"136            kwargs = {137                "TrafficMirrorFilterId": fulfillment.traffic_mirror_filter_id,138                "RuleAction": "accept",139                "Description": description,140            }141            if rule.sourcePortRange:142                sourcePortRange = self._convert_port_range_to_aws_port_range(143                    rule.sourcePortRange144                )145                kwargs["SourcePortRange"] = sourcePortRange146            if rule.destinationPortRange:147                destinationPortRange = self._convert_port_range_to_aws_port_range(148                    rule.destinationPortRange149                )150                kwargs["DestinationPortRange"] = destinationPortRange151            kwargs["TrafficDirection"] = rule.direction152            kwargs["RuleNumber"] = i153            kwargs["Protocol"] = (154                rule.protocol155                if isinstance(rule.protocol, int)156                else socket.getprotobyname(rule.protocol)157            )158            kwargs["DestinationCidrBlock"] = (159                rule.destinationCidr or self._get_destination_cidr()160            )161            kwargs["SourceCidrBlock"] = rule.sourceCidr or self._get_source_cidr()162            ec2_client.create_traffic_mirror_filter_rule(**kwargs)163    def _convert_port_range_to_aws_port_range(self, port_range):164        """# noqa165        :param cloudshell.cp.core.models.PortRange port_range:166        """167        aws_port_range = {}168        if port_range.fromPort:169            aws_port_range["FromPort"] = int(port_range.fromPort)170        if port_range.toPort:171            aws_port_range["ToPort"] = int(port_range.toPort)172        return aws_port_range173    @staticmethod174    def _get_source_cidr():175        return "0.0.0.0/0"176    @staticmethod177    def _get_destination_cidr():178        return "0.0.0.0/0"179    # endregion180    # endregion181    # region traffic mirror targets182    @staticmethod183    def create_traffic_mirror_target_from_nic(ec2_client, target_nic, tags):184        description = str(tags[0]["Value"])185        response = ec2_client.create_traffic_mirror_target(186            NetworkInterfaceId=target_nic,187            Description=description,188            TagSpecifications=[{"ResourceType": "traffic-mirror-target", "Tags": tags}],189        )190        return response["TrafficMirrorTarget"]["TrafficMirrorTargetId"]191    @staticmethod192    def find_traffic_targets_by_nics(ec2_client, target_nics):193        response = ec2_client.describe_traffic_mirror_targets(194            Filters=[{"Name": "network-interface-id", "Values": target_nics}]195        )196        return {197            x["NetworkInterfaceId"]: x["TrafficMirrorTargetId"]198            for x in response["TrafficMirrorTargets"]199        }200    @staticmethod201    def find_traffic_mirror_targets_by_reservation_id(ec2_client, reservation_id):202        traffic_mirror_targets = []203        response = ec2_client.describe_traffic_mirror_targets(204            Filters=[205                {"Name": "tag:ReservationId", "Values": [str(reservation_id)]},206            ],207            MaxResults=123,208        )209        traffic_mirror_targets.extend(response["TrafficMirrorTargets"])210        while "NextToken" in response:211            response = ec2_client.describe_traffic_mirror_filters(212                NextToken=response["NextToken"]213            )214            traffic_mirror_targets.extend(response["TrafficMirrorTargets"])215        return [target["TrafficMirrorTargetId"] for target in traffic_mirror_targets]216    @staticmethod217    def find_traffic_mirror_target_ids_by_target_nic_ids(218        ec2_client, traffic_target_nic_ids219    ):220        """# noqa221        :param list[str] traffic_target_nic_ids:222        """223        traffic_mirror_targets = []224        response = ec2_client.describe_traffic_mirror_targets(225            Filters=[226                {"Name": "network-interface-id", "Values": traffic_target_nic_ids},227            ],228            MaxResults=100,229        )230        traffic_mirror_targets.extend(response["TrafficMirrorTargets"])231        while "NextToken" in response:232            response = ec2_client.describe_traffic_mirror_filters(233                NextToken=response["NextToken"]234            )235            traffic_mirror_targets.extend(response["TrafficMirrorTargets"])236        return [target["TrafficMirrorTargetId"] for target in traffic_mirror_targets]237    @staticmethod238    def find_traffic_mirror_target_nic_id_by_target_id(239        ec2_client, traffic_mirror_target_id240    ):241        """# noqa242        :param str traffic_mirror_target_id:243        """244        traffic_mirror_targets = []245        response = ec2_client.describe_traffic_mirror_targets(246            Filters=[247                {248                    "Name": "traffic-mirror-target-id",249                    "Values": [traffic_mirror_target_id],250                },251            ],252            MaxResults=100,253        )254        traffic_mirror_targets.extend(response["TrafficMirrorTargets"])255        while "NextToken" in response:256            response = ec2_client.describe_traffic_mirror_filters(257                NextToken=response["NextToken"]258            )259            traffic_mirror_targets.extend(response["TrafficMirrorTargets"])260        return next(261            target["TrafficMirrorTargetId"] for target in traffic_mirror_targets262        )263    # endregion264    def _empty(self):...

sap_filter.py

Source:sap_filter.py

...19        return self._create_filter()20    def _find_filter(self) -> Union[str, None]:21        """ return the filter id"""22        try:23            filters = self.ec2_client.describe_traffic_mirror_filters(24                Filters=[25                    {26                        "Name": "description",  # name/name isn't supported27                        "Values": [self._filter_description],28                    }29                ]30            )["TrafficMirrorFilters"]31        except Exception as e:32            if "not found" not in str(e):33                raise34            return None35        if not filters:36            return None37        if len(filters) > 1:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.