httpx: added support for -tls_grab

secator/runners/_base.py

@@ -827,12 +827,18 @@ def _process_item(self, item: dict):
 		if not self.run_validators('item', item):
 			return None
 
-		# Run item hooks
-		item = self.run_hooks('on_item_pre_convert', item)
-		if not item:
-			return None
-
 		# Convert output dict to another schema
+		if isinstance(item, OutputType):
+			pass
+		elif isinstance(item, dict):
+			item = self.run_hooks('on_item_pre_convert', item)
+			if not item:
+				return None
+			if not self.orig:
+				item = self._convert_item_schema(item)
+			else:
+				item = DotMap(item)
+
 		if isinstance(item, dict) and not self.orig:
 			item = self._convert_item_schema(item)
 		elif isinstance(item, OutputType):

secator/tasks/httpx.py

@@ -9,8 +9,11 @@
 								 RATE_LIMIT, RETRIES, THREADS,
 								 TIMEOUT, URL, USER_AGENT)
 from secator.config import CONFIG
+from secator.output_types import Url, Subdomain
 from secator.tasks._categories import Http
-from secator.utils import sanitize_url
+from secator.utils import (sanitize_url,
+						   extract_root_domain_from_domain,
+						   remove_first_subdomain)
 
 
 @task()
@@ -35,7 +38,8 @@ class httpx(Http):
 		'screenshot': {'is_flag': True, 'short': 'ss', 'default': False, 'help': 'Screenshot response'},
 		'system_chrome': {'is_flag': True, 'default': False, 'help': 'Use local installed Chrome for screenshot'},
 		'headless_options': {'is_flag': False, 'short': 'ho', 'default': None, 'help': 'Headless Chrome additional options'},
-		'follow_host_redirects': {'is_flag': True, 'short': 'fhr', 'default': None, 'help': 'Follow redirects on the same host'}  # noqa: E501
+		'follow_host_redirects': {'is_flag': True, 'short': 'fhr', 'default': None, 'help': 'Follow redirects on the same host'},  # noqa: E501
+		'tls_grab': {'is_flag': True, 'default': False, 'help': 'Grab some informations from the tls certificate'}
 	}
 	opt_key_map = {
 		HEADER: 'header',
@@ -68,6 +72,7 @@ class httpx(Http):
 	proxy_socks5 = True
 	proxy_http = True
 	profile = 'cpu'
+	output_types = [Url, Subdomain]
 
 	@staticmethod
 	def on_init(self):
@@ -80,6 +85,7 @@ def on_init(self):
 			self.cmd += f' -srd {self.reports_folder}/.outputs'
 		if screenshot:
 			self.cmd += ' -esb -ehb'
+		self.domains = []
 
 	@staticmethod
 	def on_item_pre_convert(self, item):
@@ -94,6 +100,22 @@ def on_item_pre_convert(self, item):
 		item[URL] = item.get('final_url') or item[URL]
 		return item
 
+	@staticmethod
+	def on_json_loaded(self, item):
+		"""Extract domain from TLS certificate and yield a Subdomain if present."""
+		yield item
+		tls = item.get('tls', None)
+		if tls:
+			subject_cn = tls.get('subject_cn', None)
+			subject_an = tls.get('subject_an', [])
+			cert_domains = subject_an
+			if subject_cn:
+				cert_domains.append(subject_cn)
+			for cert_domain in cert_domains:
+				subdomain = self._create_subdomain_from_tls_cert(cert_domain)
+				if subdomain:
+					yield subdomain
+
 	@staticmethod
 	def on_end(self):
 		store_responses = self.get_opt_value('store_responses')
@@ -108,3 +130,15 @@ def on_end(self):
 				os.remove(index_spath)
 			if os.path.exists(index_spath2):
 				os.remove(index_spath2)
+
+	def _create_subdomain_from_tls_cert(self, cert_domain):
+		"""Extract subdomains from TLS certificate."""
+		if cert_domain.startswith('*'):
+			cert_domain = remove_first_subdomain(cert_domain)
+		if cert_domain in self.domains:
+			return None
+		self.domains.append(cert_domain)
+		return Subdomain(
+			host=cert_domain,
+			domain=extract_root_domain_from_domain(cert_domain)
+		)

secator/utils.py

@@ -439,3 +439,46 @@ def print_version():
 	console.print(f'[bold gold3]Lib folder[/]: {LIB_FOLDER}')
 	if status == 'outdated':
 		console.print('[bold red]secator is outdated, run "secator update" to install the latest version.')
+
+
+def extract_root_domain_from_domain(domain):
+	"""Extract the root domain from a given domain in input
+	Regex for domain validation taken and adapted from :
+	https://www.geeksforgeeks.org/how-to-validate-a-domain-name-using-regular-expression/
+	For example :
+	test.example.org will return example.org
+	Args:
+		domain (str): a domain name (like dns.google.com or wikipedia.org)
+	Returns:
+		str: The root domain.
+	"""
+	match = re.search(r"^([A-Za-z0-9-]{1,63}\.)+[A-Za-z]{2,6}$", domain)
+	if not match:
+		raise TypeError(f'Given domain {domain} is malformed')
+	domain_parts = domain.split('.')
+	if len(domain_parts) >= 2:
+		return '.'.join(domain_parts[-2:])
+	else:
+		raise TypeError(f'Fatal error this should never append, error caused by {domain}')
+
+
+def remove_first_subdomain(domain):
+	"""Remove the first subdomain
+	Regex for domain validation taken and adapted from :
+	https://www.geeksforgeeks.org/how-to-validate-a-domain-name-using-regular-expression/
+	For example :
+	test.example.org will return example.org
+	tata.toto.titi.example.org will return toto.titi.example.org
+	Args:
+		domain (str): a domain name (like dns.google.com or wikipedia.org)
+	Returns:
+		str: The domain without the first subdomain
+	"""
+	match = re.search(r"^(\*\.)?([A-Za-z0-9-]{1,63}\.)+[A-Za-z]{2,6}$", domain)
+	if not match:
+		raise TypeError(f'Given domain {domain} is malformed')
+	domain_parts = domain.split('.')
+	if len(domain_parts) >= 3:
+		return '.'.join(domain_parts[1:])
+	else:
+		raise TypeError(f'The domain {domain} does not contain a subdomain')