How to use create_traffic_mirror_filter method in localstack

Best Python code snippet using localstack_python

create_mirror_session.py

Source:create_mirror_session.py

...141    #print("Create Traffic Mirror Target response is ")142    debugprint(response)143    return(response)144    145def create_traffic_mirror_filter():146    ec2 = boto3.client('ec2')147    response = ec2.create_traffic_mirror_filter(148        Description = 'bhayes-winlogbeat-zeek-mirror-filter',149        TagSpecifications=[150            {151                'ResourceType': 'traffic-mirror-filter',152                'Tags': make_tags('winlogbeat traffic mirror filter')153            }154        ]   155    )156    debugprint(response)157    TrafficMirrorFilter = response['TrafficMirrorFilter']158    filter_id = TrafficMirrorFilter['TrafficMirrorFilterId']159    print("filter ID is " + filter_id)160    return(filter_id)161    162    163def create_traffic_mirror_filter_rule(filter_id):164    ec2 = boto3.client('ec2')165    ingress_response = ec2.create_traffic_mirror_filter_rule(166        TrafficMirrorFilterId = filter_id,167        TrafficDirection = 'ingress',168        RuleNumber = 100,169        RuleAction = 'accept',170        SourceCidrBlock = '0.0.0.0/0',171        DestinationCidrBlock = '10.100.0.0/24',172        Description = 'Default Allow All'173    )174    print("create ingress rule")175    debugprint(json.dumps(ingress_response))176    ec2 = boto3.client('ec2')177    egress_response = ec2.create_traffic_mirror_filter_rule(178        TrafficMirrorFilterId = filter_id,179        TrafficDirection = 'egress',180        RuleNumber = 100,181        RuleAction = 'accept',182        SourceCidrBlock = '10.100.0.0/24',183        DestinationCidrBlock = '0.0.0.0/0',184        Description = 'Default Allow All'           185    )186    print("create egress rule")187    debugprint(json.dumps(egress_response))188    189    190def create_traffic_mirror_session(windows_nic_id,TargetId,filter_id):191    myfigs = parse_configs()192    Owner_Name = myfigs['Owner_Name']193    Owner_Email = myfigs['Owner_Email']194    demo_name = myfigs['demo_name']    195    196    ec2 = boto3.client('ec2')197    response = ec2.create_traffic_mirror_session(198        NetworkInterfaceId = windows_nic_id,199        TrafficMirrorTargetId = TargetId,200        TrafficMirrorFilterId = filter_id,201        SessionNumber = 1,202        Description = 'bhayes winlogbeat demo mirror session',203        TagSpecifications=[204            {205                'ResourceType': 'traffic-mirror-session',206                'Tags': make_tags('winlogbeat traffic mirror session')207            }208        ]        209    )210    debugprint(response)211    return(response)212    213if args.debug:214	def debugprint(*args):215		for arg in args:216			print (arg),217		print218		print219else:220	debugprint = lambda *a: None221zeek_instance_info = get_running_instance_info('zeek')222debugprint(zeek_instance_info)223zeek_host_info = parse_instance_info(zeek_instance_info)224debugprint(json.dumps(zeek_host_info))225zeek_nic_id = zeek_host_info['NetworkInterfaceId']226windows_instance_info = get_running_instance_info('windows')227windows_host_info = parse_instance_info(windows_instance_info)228windows_nic_id = windows_host_info['NetworkInterfaceId']229debugprint(json.dumps(windows_host_info))230filter_id = create_traffic_mirror_filter()231create_traffic_mirror_filter_rule(filter_id)232mirror_target_response = create_traffic_mirror_target(zeek_nic_id)233TrafficMirrorTarget = mirror_target_response['TrafficMirrorTarget']234TargetId = TrafficMirrorTarget['TrafficMirrorTargetId']235print("Created Traffic Mirror Target ID " + TargetId)236mirror_session = create_traffic_mirror_session(windows_nic_id,TargetId,filter_id)237if mirror_session['TrafficMirrorSession']:238    TrafficMirrorSession = mirror_session['TrafficMirrorSession']239    TrafficMirrorSessionId = TrafficMirrorSession['TrafficMirrorSessionId']240    print("traffic mirror session " + TrafficMirrorSessionId + " created")241    242else:...

lambda_function.py

Source:lambda_function.py

...110        except Exception as e:111            print(e)112            time.sleep(0.05)113            114def create_traffic_mirror_filter():115    traffic_mirror_filters = ec2.describe_traffic_mirror_filters(116        Filters=[117            {118                'Name': 'description',119                'Values': [120                    'blstsecurity traffic mirror filter'121                ]122            },123        ]124    )['TrafficMirrorFilters']125    if len(traffic_mirror_filters):126        return ''127    128    tmf_id = ec2.create_traffic_mirror_filter(129        Description='blstsecurity traffic mirror filter'130    )['TrafficMirrorFilter']['TrafficMirrorFilterId']131    create_traffic_mirror_filter_rule(tmf_id, 'ingress', 100)132    create_traffic_mirror_filter_rule(tmf_id, 'egress', 100)133    return tmf_id134    135def create_traffic_mirror_session(source_network_interface_id, tmt_id, tmf_id):136    if tmf_id == '':137        return138    tms = ec2.create_traffic_mirror_session(139        NetworkInterfaceId=source_network_interface_id,140        TrafficMirrorTargetId=tmt_id,141        TrafficMirrorFilterId=tmf_id,142        SessionNumber=1,143        Description='blstsecurity traffic mirror session'144    )145    146def add_tgw_route_table(route_table_id, tgw_id):147    try:148        ec2.create_route(149            DestinationCidrBlock=blstsec_destination,150            TransitGatewayId=tgw_id,151            RouteTableId=route_table_id152        )153        return 'traffic mirroring integration installed successfully'154    except:155        return 'Wait until blstsecurity will accept the transit gateway attachment and then run the lambda again'156    157def lambda_handler(event, context):158    source_instance_data = get_instance_data()159    shared_resources_ids = get_resource_share_ids(get_resource_share_list())160    if len(shared_resources_ids):161        create_transit_gateway_attachment(shared_resources_ids['tgw'], source_instance_data['vpc_id'] , source_instance_data['subnet_id'])162        create_traffic_mirror_session(source_instance_data['network_interface_id'], shared_resources_ids['tmt'], create_traffic_mirror_filter())163        return add_tgw_route_table(source_instance_data['route_table_id'], shared_resources_ids['tgw'])164    else:...

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.