Best Python code snippet using localstack_python
sg_builder_v1.7.py
Source:sg_builder_v1.7.py
...188cidr_blk = vpc_info['Vpcs'][0]['CidrBlock']189try:190 infra_sg = conn.create_security_group(GroupName='Infrastructure-Management-SG',Description='Cloud Operations - Remote Access, AD Services, etc',VpcId=vpc_id)191 conn.create_tags(Resources=[infra_sg['GroupId']],Tags=[{'Key': 'Name', 'Value': 'Infrastructure-Management-SG'}])192 conn.authorize_security_group_ingress(GroupId=infra_sg['GroupId'],IpProtocol='tcp',FromPort=22,ToPort=22,CidrIp=cidr_blk)193 conn.authorize_security_group_ingress(GroupId=infra_sg['GroupId'],IpProtocol='tcp',FromPort=22,ToPort=22,CidrIp='172.25.10.140/32')194 conn.authorize_security_group_ingress(GroupId=infra_sg['GroupId'],IpProtocol='tcp',FromPort=22,ToPort=22,CidrIp='172.25.17.22/32')195except:196 pass197print('SGs Created!')198#####################199# Implement SG Rules:200#####################201print('Implementing Rules...') 202# Implement SG Rules:203for i in sg_lib.keys():204 for x in range(4,500):205 if sgr['C{}'.format(x)].value == None:206 continue207 else:208 sg_id = sg_lib[i]209 all_protocols = '-1'210 if sgr['F{}'.format(x)].value == 'All':211 protocol = 'All'212 else:213 protocol = sgr['F{}'.format(x)].value.lower()214 if sgr['G{}'.format(x)].value == 'All':215 port = 'All'216 elif type(sgr['G{}'.format(x)].value) == long:217 port = str(sgr['G{}'.format(x)].value)218 # Account for ranges:219 elif ' - ' in str(sgr['G{}'.format(x)].value):220 port = str(sgr['G{}'.format(x)].value)221 port = port.split(' - ')222 from_port = int(port[0])223 to_port = int(port[1])224 elif '-' in str(sgr['G{}'.format(x)].value):225 port = str(sgr['G{}'.format(x)].value)226 port = port.split('-')227 from_port = int(port[0])228 to_port = int(port[1])229 else:230 port = str(sgr['G{}'.format(x)].value)231 if sgr['I{}'.format(x)].value in sg_lib.keys():232 allowed_source = sg_lib[sgr['I{}'.format(x)].value]233 else:234 allowed_source = sgr['I{}'.format(x)].value235 allowed_source = allowed_source.replace('\n', '')236 sg_name = sgr['C{}'.format(x)].value237 direction = sgr['D{}'.format(x)].value238 if i == sg_name and direction == 'Inbound':239 if sg_name in sgr['I{}'.format(x)].value and protocol == 'All':240 try:241 conn.authorize_security_group_ingress(GroupId=sg_id,IpPermissions=[{'IpProtocol':'-1','UserIdGroupPairs':[{'GroupId':allowed_source,'VpcId':vpc_id}]}])242 except:243 pass244 elif allowed_source in sg_lib.keys() and protocol == 'All':245 try:246 conn.authorize_security_group_ingress(GroupId=sg_id,IpPermissions=[{'IpProtocol':'-1','UserIdGroupPairs':[{'GroupId':allowed_source,'VpcId':vpc_id}]}])247 except:248 pass249 250 elif protocol == 'All' and port == 'All':251 try:252 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=all_protocols,CidrIp=allowed_source)253 except:254 pass255 elif sgr['I{}'.format(x)].value in sg_lib.keys() and '-' in str(sgr['G{}'.format(x)].value):256 try:257 conn.authorize_security_group_ingress(GroupId=sg_id,IpPermissions=[{'IpProtocol':protocol,'FromPort':from_port,'ToPort':to_port,'UserIdGroupPairs':[{'GroupId':allowed_source,'VpcId':vpc_id}]}])258 except:259 pass260 elif sgr['I{}'.format(x)].value in sg_lib.keys() and ',' in str(sgr['G{}'.format(x)].value):261 port = str(sgr['G{}'.format(x)].value)262 port = port.replace(' ', '')263 port = port.split(',')264 for y in port:265 try:266 conn.authorize_security_group_ingress(GroupId=sg_id,IpPermissions=[{'IpProtocol':protocol,'FromPort':int(y),'ToPort':int(y),'UserIdGroupPairs':[{'GroupId':allowed_source,'VpcId':vpc_id}]}])267 except:268 pass269 elif sgr['I{}'.format(x)].value in sg_lib.keys():270 try:271 conn.authorize_security_group_ingress(GroupId=sg_id,IpPermissions=[{'IpProtocol':protocol,'FromPort':int(port),'ToPort':int(port),'UserIdGroupPairs':[{'GroupId':allowed_source,'VpcId':vpc_id}]}])272 except:273 pass274 elif ',' in str(sgr['G{}'.format(x)].value) and 'Hosted VPC' in sgr['I{}'.format(x)].value:275 port = str(sgr['G{}'.format(x)].value)276 port = port.replace(' ', '')277 port = port.split(',')278 for y in port:279 try:280 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=int(y),ToPort=int(y),CidrIp=cidr_blk)281 except:282 pass283 elif ',' in str(sgr['G{}'.format(x)].value):284 port = str(sgr['G{}'.format(x)].value)285 port = port.replace(' ', '')286 port = port.split(',')287 for y in port:288 try:289 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=int(y),ToPort=int(y),CidrIp=allowed_source)290 except:291 pass292 elif ' - ' in str(sgr['G{}'.format(x)].value) and 'Hosted VPC' in sgr['I{}'.format(x)].value:293 try:294 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=from_port,ToPort=to_port,CidrIp=cidr_blk)295 except:296 pass297 elif '-' in str(sgr['G{}'.format(x)].value) and 'Hosted VPC' in sgr['I{}'.format(x)].value:298 try:299 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=from_port,ToPort=to_port,CidrIp=cidr_blk)300 except:301 pass302 elif 'Hosted VPC' in sgr['I{}'.format(x)].value:303 try:304 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=int(port),ToPort=int(port),CidrIp=cidr_blk)305 except:306 pass307 elif ' - ' in str(sgr['G{}'.format(x)].value):308 try:309 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=from_port,ToPort=to_port,CidrIp=allowed_source)310 except:311 pass312 elif '-' in str(sgr['G{}'.format(x)].value):313 try:314 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=from_port,ToPort=to_port,CidrIp=allowed_source)315 except:316 pass317 elif sgr['I{}'.format(x)].value not in sg_lib.keys():318 if port == 'All':319 try:320 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=0,ToPort=65535,CidrIp=allowed_source)321 except:322 pass323 else:324 try:325 conn.authorize_security_group_ingress(GroupId=sg_id,IpProtocol=protocol,FromPort=int(port),ToPort=int(port),CidrIp=allowed_source)326 except:327 pass328 elif i == sg_name and direction == 'Outbound':329 330 if protocol == 'All' and port == 'All':331 try:332 conn.authorize_security_group_egress(GroupId=sg_id,IpProtocol=all_protocols,CidrIp=allowed_source)333 except:334 pass335 elif port == 'All':336 try:337 conn.authorize_security_group_egress(GroupId=sg_id,IpProtocol=protocol,FromPort=0,ToPort=65535,CidrIp=allowed_source)338 except:339 pass...
create_vpc.py
Source:create_vpc.py
...162 application_sec_group.create_tags(Tags=[{"Key": "Name", "Value": awsvars['applicationSecurityGroupName']}])163 alb_sec_group.create_tags(Tags=[{"Key": "Name", "Value": awsvars['albSecurityGroupName']}])164 rds_sec_group.create_tags(Tags=[{"Key": "Name", "Value": awsvars['rdsSecurityGroupName']}])165 print("Creating Security Group Rules")166 ec2client.authorize_security_group_ingress(GroupId=alb_sec_group.id,167 IpProtocol='tcp',168 FromPort=80,169 ToPort=80,170 CidrIp='0.0.0.0/0'171 )172 ec2client.authorize_security_group_ingress(GroupId=alb_sec_group.id,173 IpProtocol='tcp',174 FromPort=443,175 ToPort=443,176 CidrIp='0.0.0.0/0'177 )178 ec2client.authorize_security_group_ingress(GroupId=alb_sec_group.id,179 IpProtocol='tcp',180 FromPort=22,181 ToPort=22,182 CidrIp=awsvars['sshCidrBlock1']183 )184 ec2client.authorize_security_group_ingress(GroupId=alb_sec_group.id,185 IpProtocol='tcp',186 FromPort=22,187 ToPort=22,188 CidrIp=awsvars['sshCidrBlock2']189 )190 ec2client.authorize_security_group_ingress(GroupId=alb_sec_group.id,191 IpProtocol='tcp',192 FromPort=22,193 ToPort=22,194 CidrIp=awsvars['sshCidrBlock3']195 )196 # Allow Application Security Group to receive traffic from ALB Security group197 ec2client.authorize_security_group_ingress(GroupId=application_sec_group.id,198 IpPermissions=[{'IpProtocol': 'tcp',199 'FromPort': 80,200 'ToPort': 80,201 'UserIdGroupPairs': [{'GroupId': alb_sec_group.id}]202 }]203 )204 ec2client.authorize_security_group_ingress(GroupId=application_sec_group.id,205 IpPermissions=[{'IpProtocol': 'tcp',206 'FromPort': 22,207 'ToPort': 22,208 'UserIdGroupPairs': [{'GroupId': alb_sec_group.id}]209 }]210 )211 # Allow RDS Security Group to receive traffic from Application Security group on 3306212 ec2client.authorize_security_group_ingress(GroupId=rds_sec_group.id,213 IpPermissions=[{'IpProtocol': 'tcp',214 'FromPort': 3306,215 'ToPort': 3306,216 'UserIdGroupPairs': [217 {'GroupId': application_sec_group.id}]218 }]219 )...
ec2.py
Source:ec2.py
...45 # response = client.authorize_security_group_egress()46 raise BotorException("Modifying egress rules is not yet supported by Grouper")47 else:48 if rule.cidr:49 client.authorize_security_group_ingress(50 GroupId=group.aws_group_id,51 IpProtocol=rule.protocol,52 FromPort=rule.from_port,53 ToPort=rule.to_port,54 CidrIp=rule.cidr55 )56 else:57 client.authorize_security_group_ingress(58 GroupId=group.aws_group_id,59 IpProtocol=rule.protocol,60 FromPort=rule.from_port,61 ToPort=rule.to_port,62 SourceSecurityGroupName=rule.source_security_group.name,63 # SourceSecurityGroupOwnerId=rule.source_security_group.account.aws_account_id64 )65@sts_conn('ec2')66@rate_limited()67def revoke_rule(rule, group, account_number=None, region=None, assume_role=None, client=None):68 if rule.direction == 'egress':69 # response = client.authorize_security_group_egress()70 raise BotorException("Modifying egress rules is not yet supported by Grouper")71 else:72 client.revoke_security_group_ingress(73 GroupId=group.aws_group_id,74 IpProtocol=rule.protocol,75 FromPort=rule.from_port,76 ToPort=rule.to_port,77 CidrIp=rule.cidr,78 )79@sts_conn('ec2')80@rate_limited()81def add_groups_to_instance(instance_id, groups, account_number=None, region=None, assume_role=None, client=None):82 client.modify_instance_attribute(InstanceId=instance_id, Groups=groups)83@sts_conn('ec2')84@rate_limited()85def describe_instances(**kwargs):86 return kwargs.pop('client').get_paginator('describe_instances').paginate()87@sts_conn('ec2')88@rate_limited()89def describe_security_groups(**kwargs):90 return kwargs.pop('client').describe_security_groups(**kwargs)91@sts_conn('ec2')92@rate_limited()93def create_security_group(**kwargs):94 return kwargs.pop('client').create_security_group(**kwargs)95@sts_conn('ec2')96@rate_limited()97def authorize_security_group_ingress(**kwargs):98 return kwargs.pop('client').authorize_security_group_ingress(**kwargs)99@sts_conn('ec2')100@rate_limited()101def authorize_security_group_egress(**kwargs):...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!