How to use get_mapping method in tempest

Best Python code snippet using tempest_python

web.py

Source:web.py

...12cat = 'global'13percentageColors = c.get_percentage(c.compare_data(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))), c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))14percentageColorful = c.get_percentage(c.compare_data(c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))), c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))))15percentageFonts = c.get_percentage(c.compare_data(c.get_items_css_array('fonts', True, timestamp, cat)), c.get_items_css_array('fonts', True, timestamp, cat))16mapping_resp = c.get_mapping(c.compare_data(c.get_items_css('responsive', False, timestamp, cat)))17mapping_webserver = c.get_mapping(c.compare_data(c.get_items_tech_array('web_servers', False, timestamp, cat)))18mapping_js_frameworks = c.get_mapping(c.compare_data(c.get_items_tech_array('js_frameworks', False, timestamp, cat)))19mapping_programming_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('programming_lng', False, timestamp, cat)))20mapping_markup_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('markup_lang', False, timestamp, cat)))21mapping_site_elem = c.get_mapping(c.compare_data(c.get_items_tech_array('site_elements', False, timestamp, cat)))22mapping_font_scripts= c.get_mapping(c.compare_data(c.get_items_tech_array('font_scripts', False, timestamp, cat)))23mapping_encodings= c.get_mapping(c.compare_data(c.get_items_tech_array('character_encoding', False, timestamp, cat)))24mapping_img_formats= c.get_mapping(c.compare_data(c.get_items_tech_array('img_formats', False, timestamp, cat)))25mapping_analysis_tool = c.get_mapping(c.compare_data(c.get_items_tech_array('analysis_tool', False, timestamp, cat)))26# ALEXA AUSTRIA TOP 10027cat = 'AUT'28at_percentageColors = c.get_percentage(c.compare_data(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))), c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))29at_percentageColorful = c.get_percentage(c.compare_data(c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))), c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))))30at_percentageFonts = c.get_percentage(c.compare_data(c.get_items_css_array('fonts', True, timestamp, cat)), c.get_items_css_array('fonts', True, timestamp, cat))31at_mapping_resp = c.get_mapping(c.compare_data(c.get_items_css('responsive', False, timestamp, cat)))32at_mapping_webserver = c.get_mapping(c.compare_data(c.get_items_tech_array('web_servers', False, timestamp, cat)))33at_mapping_js_frameworks = c.get_mapping(c.compare_data(c.get_items_tech_array('js_frameworks', False, timestamp, cat)))34at_mapping_programming_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('programming_lng', False, timestamp, cat)))35at_mapping_markup_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('markup_lang', False, timestamp, cat)))36at_mapping_site_elem = c.get_mapping(c.compare_data(c.get_items_tech_array('site_elements', False, timestamp, cat)))37at_mapping_font_scripts= c.get_mapping(c.compare_data(c.get_items_tech_array('font_scripts', False, timestamp, cat)))38at_mapping_encodings= c.get_mapping(c.compare_data(c.get_items_tech_array('character_encoding', False, timestamp, cat)))39at_mapping_img_formats= c.get_mapping(c.compare_data(c.get_items_tech_array('img_formats', False, timestamp, cat)))40at_mapping_analysis_tool = c.get_mapping(c.compare_data(c.get_items_tech_array('analysis_tool', False, timestamp, cat)))41# ALEXA DESIGN TOP 10042cat = 'design'43design_percentageColors = c.get_percentage(c.compare_data(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))), c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))44design_percentageColorful = c.get_percentage(c.compare_data(c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat)))), c.color_not_neutral(c.combine_lists(c.get_items_css_array('colors', False, timestamp, cat),c.get_items_css_array('bg_colors', False, timestamp, cat))))45design_percentageFonts = c.get_percentage(c.compare_data(c.get_items_css_array('fonts', True, timestamp, cat)), c.get_items_css_array('fonts', True, timestamp, cat))46design_mapping_resp = c.get_mapping(c.compare_data(c.get_items_css('responsive', False, timestamp, cat)))47design_mapping_webserver = c.get_mapping(c.compare_data(c.get_items_tech_array('web_servers', False, timestamp, cat)))48design_mapping_js_frameworks = c.get_mapping(c.compare_data(c.get_items_tech_array('js_frameworks', False, timestamp, cat)))49design_mapping_programming_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('programming_lng', False, timestamp, cat)))50design_mapping_markup_lang = c.get_mapping(c.compare_data(c.get_items_tech_array('markup_lang', False, timestamp, cat)))51design_mapping_site_elem = c.get_mapping(c.compare_data(c.get_items_tech_array('site_elements', False, timestamp, cat)))52design_mapping_font_scripts= c.get_mapping(c.compare_data(c.get_items_tech_array('font_scripts', False, timestamp, cat)))53design_mapping_encodings= c.get_mapping(c.compare_data(c.get_items_tech_array('character_encoding', False, timestamp, cat)))54design_mapping_img_formats= c.get_mapping(c.compare_data(c.get_items_tech_array('img_formats', False, timestamp, cat)))55design_mapping_analysis_tool = c.get_mapping(c.compare_data(c.get_items_tech_array('analysis_tool', False, timestamp, cat)))56@app.route('/')57def index(fonts=percentageFonts, colors=percentageColors, responsive=mapping_resp,58                programming_lng=mapping_programming_lang,59                js_frameworks=mapping_js_frameworks, web_servers=mapping_webserver,60                font_scripts=mapping_font_scripts,61                encoding=mapping_encodings, site_elem=mapping_site_elem,62                analysis_tool=mapping_analysis_tool, markup_lang=mapping_markup_lang, img_formats=mapping_img_formats, colorful=percentageColorful,63                url_title='Top 100 Alex Websites GLOBAL'):64    return render_template('index.html', fonts=fonts, colors=colors, responsive=responsive,65                           programming_lng=programming_lng,66                           js_frameworks=js_frameworks, web_servers=web_servers,67                           font_scripts=font_scripts, encoding=encoding, site_elem=site_elem,68                           analysis_tool=analysis_tool, markup_lang=markup_lang, img_formats=img_formats,69                           colorful=colorful, url_title=url_title)...

services.py

Source:services.py

...3940		self.alert.init(['Date / Time (UTC)', 'Computer Name', 'Case', 'Channel', 'EventID', 'Description (short)', 'Service Name', 'SID', 'Payload Analysis', 'IP', 'Port', 'Payload (Raw)', 'Payload (Decoded)'])41		42		# Powershell Exec evidence43		#services = MultiMatch(query='Windows PowerShell', fields=[self.get_mapping('evt_channel_field_k')]) & (MultiMatch(query='-noni') | MultiMatch(query='-nop -w hidden') | MultiMatch(query='COMSPEC') )44		services = MultiMatch(query='Windows PowerShell', fields=[self.get_mapping('evt_channel_field_k')]) & ( blacklist_filter )45		if self.filter:46			services &= self.filter	47		self.search =self.search.query(services)48		self.resp = self.search.execute()4950		for hit in self.search.scan():51			d_hit = hit.to_dict()52			# Generic fields53			computer = self.get_value(d_hit, self.get_mapping('evt_system_field'))54			timestamp = self.get_value(d_hit, self.get_mapping('evt_time_field'))55			event_id = self.get_value(d_hit, self.get_mapping('evt_event_id_field'))56			desc = self.get_value(d_hit, self.get_mapping('evt_desc_field'))57			case = self.get_value(d_hit, self.get_mapping('case_field'))5859			channel = self.get_value(d_hit, self.get_mapping('evt_channel_field'))60			sid = '-'61			if event_id == 7045:62				continue6364			# Extract payload from Powershell raw data65			logging.debug(event_id)66			logging.debug(hit.message)67			logging.debug(self.get_value(d_hit, self.get_mapping('evt_powershell_rawdata_field')))68			logging.debug([x for x in self.get_value(d_hit, self.get_mapping('evt_powershell_rawdata_field')).split('\n') if 'HostApplication=' in x])69			logging.debug('===================================')70			payload = [x for x in self.get_value(d_hit, self.get_mapping('evt_powershell_rawdata_field')).split('\n') if 'HostApplication=' in x]71			if not payload:72				[x for x in self.get_value(d_hit, self.get_mapping('evt_powershell_rawdata_2_field')).split('\n') if 'HostApplication=' in x]73			_payload = ''74			if payload:75				_payload = payload[0].strip().replace('HostApplication=', '')7677			# TODO: decode paylaod78			mess, ip, port = 'unknown', '', ''79			is_decoded, decoded_payload  = '', ''8081			self.alert.add_alert([timestamp, computer, case, channel, event_id, desc, '-', sid, mess, ip, port, _payload, decoded_payload])8283		# Service Creation evidence84		services = MultiMatch(query=7045, fields=[self.get_mapping('evt_event_id_field')]) & ( blacklist_filter )85		search = Search(using=self.client, index=self.index)86		if self.filter:87			services &= self.filter8889		search = search.query(services)9091		resp = search.execute()92		for hit in search.scan():93			d_hit = hit.to_dict()94			# Generic fields95			computer = self.get_value(d_hit, self.get_mapping('evt_system_field'))96			timestamp = self.get_value(d_hit, self.get_mapping('evt_time_field'))97			event_id = self.get_value(d_hit, self.get_mapping('evt_event_id_field'))98			desc = self.get_value(d_hit, self.get_mapping('evt_desc_field'))99			case = self.get_value(d_hit, self.get_mapping('case_field'))100			channel = self.get_value(d_hit, self.get_mapping('evt_channel_field'))101102			sid = self.get_value(d_hit, self.get_mapping('evt_user_sid_field'))103			servicename = self.get_value(d_hit, self.get_mapping('evt_service_name_field'))104			payload = self.get_value(d_hit, self.get_mapping('evt_service_path_field'))105106			mess, ip, port = 'unknown', '', ''107			decoded_payload = ''108			# is_decoded, decoded_payload = decode_powershell(payload)109			# if is_decoded:110			# 	mess, ip, port = analyze_payload(decoded_payload)111112			self.alert.add_alert([timestamp, computer, case, channel, event_id, desc,servicename, sid,  mess, ip, port, payload, decoded_payload])113114		# 4697115		services = MultiMatch(query='4697', fields=[self.get_mapping('evt_event_id_field')]) & ( blacklist_filter )116		search = Search(using=self.client, index=self.index)117		if self.filter:118			services &= self.filter119		search = search.query(services)120		121		resp = search.execute()122		for hit in search.scan():123			d_hit = hit.to_dict()124			# Generic fields125			computer = self.get_value(d_hit, self.get_mapping('evt_system_field'))126			timestamp = self.get_value(d_hit, self.get_mapping('evt_time_field'))127			event_id = self.get_value(d_hit, self.get_mapping('evt_event_id_field'))128			desc = self.get_value(d_hit, self.get_mapping('evt_desc_field'))129			case = self.get_value(d_hit, self.get_mapping('case_field'))130			channel = self.get_value(d_hit, self.get_mapping('evt_channel_field'))131			sid = self.get_value(d_hit, self.get_mapping('evt_service_sid'))132133			servicename = self.get_value(d_hit, self.get_mapping('evt_service_name_field'))134			payload = self.get_value(d_hit, self.get_mapping('evt_service_filename_field'))135136			mess, ip, port = 'unknown', '', ''137			is_decoded, decoded_payload = decode_powershell(payload)138			if is_decoded:139				mess, ip, port = analyze_payload(decoded_payload)140141			self.alert.add_alert([timestamp, computer, case, channel, eventid, desc,servicename, sid,  mess, ip, port, payload, decoded_payload])142	143"""144	Extract BITS (Background Intelligent Transfer Service) URLs145		Event 59 - BITS started the BITS Transfer transfer job146		Event 60 - BITS stopped transferring the BITS Transfer transfer job 147		Event 61 - BITS stopped transferring the BITS Transfer transfer job wth error code148"""149class BITSService(ElasticScenario):150	def __init__(self):151		super(BITSService, self).__init__()152153	def process(self):154		bits_service = (MultiMatch(query='59', fields=[self.get_mapping('evt_event_id_field')]) | MultiMatch(query='60', fields=self.get_mapping('evt_event_id_field')) | MultiMatch(query='61', fields=self.get_mapping('evt_event_id_field'))) \155			& MultiMatch(query='Microsoft-Windows-Bits-Client/Operational', fields=self.get_mapping('evt_channel_field_k'))156157		if self.filter:158			bits_service &= self.filter159160		self.search = self.search.query(bits_service)161162		self.search.aggs.bucket('computer', 'terms', field=self.get_mapping('evt_system_field_k'), size = self.bucket_size)\163			.bucket('bits', 'terms', field=self.get_mapping('evt_bits_url_field_k'), size = self.bucket_size)164165		whitelist = self.get_conf('whitelist', default=[])166167		self.alert.init(['Computer Name', 'Location', 'Path', 'Params', 'Query', 'Nb Hits'])168		self.resp = self.search.execute()169		for computer_data in self.resp.aggregations.computer:170			for bits_data in computer_data.bits:171				url = urlparse(bits_data.key)172				# Check if in whitelist173				matches = [domain for domain in whitelist if domain in url.netloc]174				if not matches:175					self.alert.add_alert([computer_data.key, '{}://{}'.format(url.scheme, url.netloc) if url.netloc else url.geturl(), url.path, url.params, url.query, bits_data.doc_count])
...

auctions_server_data.py

Source:auctions_server_data.py

1from mock import sentinel2AUCTION_DOC_ID = 'some_id'3proxy_data_proxy_path = \4    {'server_config_redis': sentinel.REDIS,5     'connection_limit': sentinel.connection_limit,6     'proxy_connection_pool': sentinel.proxy_pool,7     'get_mapping': sentinel.get_mapping,8     'proxy_path': sentinel.proxy_path,9     'stream_proxy': sentinel.stream_proxy,10     'event_sources_pool': sentinel.event_sources_pool}11# there is no '/login' in request_url12# there is 'X-Forwarded-For' in headers13proxy_data_no_proxy_path_forwarded_header_1 = \14    {'server_config_redis': sentinel.REDIS,15     'get_mapping': sentinel.get_mapping,16     'proxy_path': None,17     'db': ['some_id0', AUCTION_DOC_ID, 'some_id1'],18     'request_headers': {19         'h0': 'h0',20         'X-Forwarded-For': 'hx',21         'h1': 'h1',22         'Host': 'new_host'23     },24     'request_url': 'http://netloc/path;params?query=argument#fragment',25     'transformed_url': 'http://new_host/path;params?query=argument#fragment',26     'redirect_url': sentinel.redirect_url}27# there is '/login' in request_url28# there is 'X-Forwarded-For' in headers29proxy_data_no_proxy_path_forwarded_header_2 = \30    {'server_config_redis': sentinel.REDIS,31     'get_mapping': sentinel.get_mapping,32     'proxy_path': None,33     'db': ['some_id0', AUCTION_DOC_ID, 'some_id1'],34     'request_headers': {35         'h0': 'h0',36         'X-Forwarded-For': 'hx',37         'h1': 'h1',38         'Host': 'new_host'39     },40     'request_url': 'http://netloc/login/path;params?query=argument#fragment',41     'transformed_url': 'http://new_host/path;params?query=argument#fragment',42     'redirect_url': sentinel.redirect_url}43# there is no 'X-Forwarded-For' in headers44proxy_data_no_proxy_path_no_forwarded_header = \45    {'server_config_redis': sentinel.REDIS,46     'get_mapping': sentinel.get_mapping,47     'proxy_path': None,48     'db': ['some_id0', AUCTION_DOC_ID, 'some_id1'],49     'request_headers': {'h0': 'h0',50                         'h1': 'h1',51                         'Host': 'new_host'},52     'abort': sentinel.abort}53proxy_data_no_proxy_path = \54    {'server_config_redis': sentinel.REDIS,55     'get_mapping': sentinel.get_mapping,56     'proxy_path': None,57     'pysse': sentinel.pysse,...

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.