path: root/archinstall/lib/profiles.py

import os, urllib.request, urllib.parse, ssl, json, re
import importlib.util, sys, glob, hashlib, logging
from collections import OrderedDict
from .general import multisplit, sys_command
from .exceptions import *
from .networking import *
from .output import log
from .storage import storage

def grab_url_data(path):
	safe_path = path[:path.find(':')+1]+''.join([item if item in ('/', '?', '=', '&') else urllib.parse.quote(item) for item in multisplit(path[path.find(':')+1:], ('/', '?', '=', '&'))])
	ssl_context = ssl.create_default_context()
	ssl_context.check_hostname = False
	ssl_context.verify_mode=ssl.CERT_NONE
	response = urllib.request.urlopen(safe_path, context=ssl_context)
	return response.read()

def list_profiles(filter_irrelevant_macs=True, subpath='', filter_top_level_profiles=False):
	# TODO: Grab from github page as well, not just local static files
	if filter_irrelevant_macs:
		local_macs = list_interfaces()

	cache = {}
	# Grab all local profiles found in PROFILE_PATH
	for PATH_ITEM in storage['PROFILE_PATH']:
		for root, folders, files in os.walk(os.path.abspath(os.path.expanduser(PATH_ITEM+subpath))):
			for file in files:
				if os.path.splitext(file)[1] == '.py':
					tailored = False
					if len(mac := re.findall('(([a-zA-z0-9]{2}[-:]){5}([a-zA-z0-9]{2}))', file)):
						if filter_irrelevant_macs and mac[0][0].lower() not in local_macs:
							continue
						tailored = True

					description = ''
					with open(os.path.join(root, file), 'r') as fh:
						first_line = fh.readline()
						if first_line[0] == '#':
							description = first_line[1:].strip()

					cache[file[:-3]] = {'path' : os.path.join(root, file), 'description' : description, 'tailored' : tailored}
			break

	# Grab profiles from upstream URL
	if storage['PROFILE_DB']:
		profiles_url = os.path.join(storage["UPSTREAM_URL"]+subpath, storage['PROFILE_DB'])
		try:
			profile_list = json.loads(grab_url_data(profiles_url))
		except urllib.error.HTTPError as err:
			print(f'Error: Listing profiles on URL "{profiles_url}" resulted in:', err)
			return cache
		except:
			print(f'Error: Could not decode "{profiles_url}" result as JSON:', err)
			return cache
		
		for profile in profile_list:
			if os.path.splitext(profile)[1] == '.py':
				tailored = False
				if len(mac := re.findall('(([a-zA-z0-9]{2}[-:]){5}([a-zA-z0-9]{2}))', profile)):
					if filter_irrelevant_macs and mac[0][0].lower() not in local_macs:
						continue
					tailored = True

				cache[profile[:-3]] = {'path' : os.path.join(storage["UPSTREAM_URL"]+subpath, profile), 'description' : profile_list[profile], 'tailored' : tailored}

	if filter_top_level_profiles:
		for profile in list(cache.keys()):
			if Profile(None, profile).is_top_level_profile() is False:
				del(cache[profile])

	return cache

class Script():
	def __init__(self, profile, installer=None):
		# profile: https://hvornum.se/something.py
		# profile: desktop
		# profile: /path/to/profile.py
		self.profile = profile
		self.installer = installer
		self.converted_path = None
		self.spec = None
		self.examples = None
		self.namespace = os.path.splitext(os.path.basename(self.path))[0]
		self.original_namespace = self.namespace
		log(f"Script {self} has been loaded with namespace '{self.namespace}'", level=logging.DEBUG)

	def __enter__(self, *args, **kwargs):
		self.execute()
		return sys.modules[self.namespace]

	def __exit__(self, *args, **kwargs):
		# TODO: https://stackoverflow.com/questions/28157929/how-to-safely-handle-an-exception-inside-a-context-manager
		if len(args) >= 2 and args[1]:
			raise args[1]

	def localize_path(self, profile_path):
		if (url := urllib.parse.urlparse(profile_path)).scheme and url.scheme in ('https', 'http'):
			if not self.converted_path:
				self.converted_path = f"/tmp/{os.path.basename(self.profile).replace('.py', '')}_{hashlib.md5(os.urandom(12)).hexdigest()}.py"

				with open(self.converted_path, "w") as temp_file:
					temp_file.write(urllib.request.urlopen(url.geturl()).read().decode('utf-8'))

			return self.converted_path
		else:
			return profile_path

	@property
	def path(self):
		parsed_url = urllib.parse.urlparse(self.profile)

		# The Profile was not a direct match on a remote URL
		if not parsed_url.scheme:
			# Try to locate all local or known URL's
			if not self.examples:
				self.examples = list_profiles()

			if f"{self.profile}" in self.examples:
				return self.localize_path(self.examples[self.profile]['path'])
			# TODO: Redundant, the below block shouldn't be needed as profiles are stripped of their .py, but just in case for now:
			elif f"{self.profile}.py" in self.examples:
				return self.localize_path(self.examples[f"{self.profile}.py"]['path'])

			# Path was not found in any known examples, check if it's an absolute path
			if os.path.isfile(self.profile):
				return self.profile

			raise ProfileNotFound(f"File {self.profile} does not exist in {storage['PROFILE_PATH']}")
		elif parsed_url.scheme in ('https', 'http'):
			return self.localize_path(self.profile)
		else:
			raise ProfileNotFound(f"Cannot handle scheme {parsed_url.scheme}")

	def load_instructions(self, namespace=None):
		if namespace:
			self.namespace = namespace

		self.spec = importlib.util.spec_from_file_location(self.namespace, self.path)
		imported = importlib.util.module_from_spec(self.spec)
		sys.modules[self.namespace] = imported

		return self

	def execute(self):
		if not self.namespace in sys.modules or self.spec is None:
			self.load_instructions()

		self.spec.loader.exec_module(sys.modules[self.namespace])

		return sys.modules[self.namespace]

class Profile(Script):
	def __init__(self, installer, path, args={}):
		super(Profile, self).__init__(path, installer)

	def __dump__(self, *args, **kwargs):
		return {'path' : self.path}

	def __repr__(self, *args, **kwargs):
		return f'Profile({os.path.basename(self.profile)})'

	def install(self):
		# Before installing, revert any temporary changes to the namespace.
		# This ensures that the namespace during installation is the original initiation namespace.
		# (For instance awesome instead of aweosme.py or app-awesome.py)
		self.namespace = self.original_namespace
		return self.execute()

	def has_prep_function(self):
		with open(self.path, 'r') as source:
			source_data = source.read()

			# Some crude safety checks, make sure the imported profile has
			# a __name__ check and if so, check if it's got a _prep_function()
			# we can call to ask for more user input.
			#
			# If the requirements are met, import with .py in the namespace to not
			# trigger a traditional:
			#     if __name__ == 'moduleName'
			if '__name__' in source_data and '_prep_function' in source_data:
				with self.load_instructions(namespace=f"{self.namespace}.py") as imported:
					if hasattr(imported, '_prep_function'):
						return True
		return False

	def has_post_install(self):
		with open(self.path, 'r') as source:
			source_data = source.read()

			# Some crude safety checks, make sure the imported profile has
			# a __name__ check and if so, check if it's got a _prep_function()
			# we can call to ask for more user input.
			#
			# If the requirements are met, import with .py in the namespace to not
			# trigger a traditional:
			#     if __name__ == 'moduleName'
			if '__name__' in source_data and '_post_install' in source_data:
				with self.load_instructions(namespace=f"{self.namespace}.py") as imported:
					if hasattr(imported, '_post_install'):
						return True

	def is_top_level_profile(self):
		with open(self.path, 'r') as source:
			source_data = source.read()

			# TODO: I imagine that there is probably a better way to write this.
			return 'top_level_profile = True' in source_data

	@property
	def packages(self) -> list:
		"""
		Returns a list of packages baked into the profile definition.
		If no package definition has been done, .packages() will return None.
		"""
		with open(self.path, 'r') as source:
			source_data = source.read()

			# Some crude safety checks, make sure the imported profile has
			# a __name__ check before importing.
			#
			# If the requirements are met, import with .py in the namespace to not
			# trigger a traditional:
			#     if __name__ == 'moduleName'
			if '__name__' in source_data and '__packages__' in source_data:
				with self.load_instructions(namespace=f"{self.namespace}.py") as imported:
					if hasattr(imported, '__packages__'):
						return imported.__packages__
		return None
	

	def has_post_install(self):
		with open(self.path, 'r') as source:
			source_data = source.read()

			# Some crude safety checks, make sure the imported profile has
			# a __name__ check and if so, check if it's got a _prep_function()
			# we can call to ask for more user input.
			#
			# If the requirements are met, import with .py in the namespace to not
			# trigger a traditional:
			#     if __name__ == 'moduleName'
			if '__name__' in source_data and '_post_install' in source_data:
				with self.load_instructions(namespace=f"{self.namespace}.py") as imported:
					if hasattr(imported, '_post_install'):
						return True

	def is_top_level_profile(self):
		with open(self.path, 'r') as source:
			source_data = source.read()
			return 'top_level_profile = True' in source_data

	@property
	def packages(self) -> list:
		"""
		Returns a list of packages baked into the profile definition.
		If no package definition has been done, .packages() will return None.
		"""
		with open(self.path, 'r') as source:
			source_data = source.read()

			# Some crude safety checks, make sure the imported profile has
			# a __name__ check before importing.
			#
			# If the requirements are met, import with .py in the namespace to not
			# trigger a traditional:
			#     if __name__ == 'moduleName'
			if '__name__' in source_data and '__packages__' in source_data:
				with self.load_instructions(namespace=f"{self.namespace}.py") as imported:
					if hasattr(imported, '__packages__'):
						return imported.__packages__
		return None
	

class Application(Profile):
	def __repr__(self, *args, **kwargs):
		return f'Application({os.path.basename(self.profile)})'

	@property
	def path(self):
		parsed_url = urllib.parse.urlparse(self.profile)

		# The Profile was not a direct match on a remote URL
		if not parsed_url.scheme:
			# Try to locate all local or known URL's
			if not self.examples:
				self.examples = list_profiles(subpath='/applications')

			if f"{self.profile}" in self.examples:
				return self.localize_path(self.examples[self.profile]['path'])
			# TODO: Redundant, the below block shouldn't be needed as profiles are stripped of their .py, but just in case for now:
			elif f"{self.profile}.py" in self.examples:
				return self.localize_path(self.examples[f"{self.profile}.py"]['path'])

			# Path was not found in any known examples, check if it's an absolute path
			if os.path.isfile(self.profile):
				return os.path.basename(self.profile)

			raise ProfileNotFound(f"Application file {self.profile} does not exist in {storage['PROFILE_PATH']}")
		elif parsed_url.scheme in ('https', 'http'):
			return self.localize_path(self.profile)
		else:
			raise ProfileNotFound(f"Application cannot handle scheme {parsed_url.scheme}")

	def install(self):
		# Before installing, revert any temporary changes to the namespace.
		# This ensures that the namespace during installation is the original initiation namespace.
		# (For instance awesome instead of aweosme.py or app-awesome.py)
		self.namespace = self.original_namespace
		return self.execute()