pym/portage/package/ebuild/_config/LicenseManager.py

# Copyright 2010 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2

__all__ = (
	'LicenseManager',
)

from portage import os
from portage.dep import ExtendedAtomDict, use_reduce
from portage.exception import InvalidDependString
from portage.localization import _
from portage.util import grabdict, grabdict_package, writemsg
from portage.versions import cpv_getkey

from portage.package.ebuild._config.helper import ordered_by_atom_specificity


class LicenseManager(object):

	def __init__(self, license_group_locations, abs_user_config, user_config=True):

		self._accept_license_str = None
		self._accept_license = None
		self._license_groups = {}
		self._plicensedict = ExtendedAtomDict(dict)
		self._undef_lic_groups = set()

		if user_config:
			license_group_locations = list(license_group_locations) + [abs_user_config]

		self._read_license_groups(license_group_locations)

		if user_config:
			self._read_user_config(abs_user_config)

	def _read_user_config(self, abs_user_config):
		licdict = grabdict_package(os.path.join(
			abs_user_config, "package.license"), recursive=1, allow_wildcard=True, allow_repo=True, verify_eapi=False)
		for k, v in licdict.items():
			self._plicensedict.setdefault(k.cp, {})[k] = \
				self.expandLicenseTokens(v)

	def _read_license_groups(self, locations):
		for loc in locations:
			for k, v in grabdict(
				os.path.join(loc, "license_groups")).items():
				self._license_groups.setdefault(k, []).extend(v)

		for k, v in self._license_groups.items():
			self._license_groups[k] = frozenset(v)

	def extract_global_changes(self, old=""):
		ret = old
		atom_license_map = self._plicensedict.get("*/*")
		if atom_license_map is not None:
			v = atom_license_map.pop("*/*", None)
			if v is not None:
				ret = " ".join(v)
				if old:
					ret = old + " " + ret
				if not atom_license_map:
					#No tokens left in atom_license_map, remove it.
					del self._plicensedict["*/*"]
		return ret
	
	def expandLicenseTokens(self, tokens):
		""" Take a token from ACCEPT_LICENSE or package.license and expand it
		if it's a group token (indicated by @) or just return it if it's not a
		group.  If a group is negated then negate all group elements."""
		expanded_tokens = []
		for x in tokens:
			expanded_tokens.extend(self._expandLicenseToken(x, None))
		return expanded_tokens

	def _expandLicenseToken(self, token, traversed_groups):
		negate = False
		rValue = []
		if token.startswith("-"):
			negate = True
			license_name = token[1:]
		else:
			license_name = token
		if not license_name.startswith("@"):
			rValue.append(token)
			return rValue
		group_name = license_name[1:]
		if traversed_groups is None:
			traversed_groups = set()
		license_group = self._license_groups.get(group_name)
		if group_name in traversed_groups:
			writemsg(_("Circular license group reference"
				" detected in '%s'\n") % group_name, noiselevel=-1)
			rValue.append("@"+group_name)
		elif license_group:
			traversed_groups.add(group_name)
			for l in license_group:
				if l.startswith("-"):
					writemsg(_("Skipping invalid element %s"
						" in license group '%s'\n") % (l, group_name),
						noiselevel=-1)
				else:
					rValue.extend(self._expandLicenseToken(l, traversed_groups))
		else:
			if self._license_groups and \
				group_name not in self._undef_lic_groups:
				self._undef_lic_groups.add(group_name)
				writemsg(_("Undefined license group '%s'\n") % group_name,
					noiselevel=-1)
			rValue.append("@"+group_name)
		if negate:
			rValue = ["-" + token for token in rValue]
		return rValue

	def _getPkgAcceptLicense(self, cpv, slot, repo):
		"""
		Get an ACCEPT_LICENSE list, accounting for package.license.
		"""
		accept_license = self._accept_license
		cp = cpv_getkey(cpv)
		cpdict = self._plicensedict.get(cp)
		if cpdict:
			cpv_slot = "%s:%s" % (cpv, slot)
			plicence_list = ordered_by_atom_specificity(cpdict, cpv_slot, repo)
			if plicence_list:
				accept_license = list(self._accept_license)
				for x in plicence_list:
					accept_license.extend(x)
		return accept_license

	def get_prunned_accept_license(self, cpv, use, lic, slot, repo):
		"""
		Generate a pruned version of ACCEPT_LICENSE, by intersection with
		LICENSE. This is required since otherwise ACCEPT_LICENSE might be
		too big (bigger than ARG_MAX), causing execve() calls to fail with
		E2BIG errors as in bug #262647.
		"""
		try:
			licenses = set(use_reduce(lic, uselist=use, flat=True))
		except InvalidDependString:
			licenses = set()
		licenses.discard('||')

		accept_license = self._getPkgAcceptLicense(cpv, slot, repo)

		if accept_license:
			acceptable_licenses = set()
			for x in accept_license:
				if x == '*':
					acceptable_licenses.update(licenses)
				elif x == '-*':
					acceptable_licenses.clear()
				elif x[:1] == '-':
					acceptable_licenses.discard(x[1:])
				elif x in licenses:
					acceptable_licenses.add(x)

			licenses = acceptable_licenses
		return ' '.join(sorted(licenses))

	def getMissingLicenses(self, cpv, use, lic, slot, repo):
		"""
		Take a LICENSE string and return a list of any licenses that the user
		may need to accept for the given package.  The returned list will not
		contain any licenses that have already been accepted.  This method
		can throw an InvalidDependString exception.

		@param cpv: The package name (for package.license support)
		@type cpv: String
		@param use: "USE" from the cpv's metadata
		@type use: String
		@param lic: "LICENSE" from the cpv's metadata
		@type lic: String
		@param slot: "SLOT" from the cpv's metadata
		@type slot: String
		@rtype: List
		@return: A list of licenses that have not been accepted.
		"""

		licenses = set(use_reduce(lic, matchall=1, flat=True))
		licenses.discard('||')

		acceptable_licenses = set()
		for x in self._getPkgAcceptLicense(cpv, slot, repo):
			if x == '*':
				acceptable_licenses.update(licenses)
			elif x == '-*':
				acceptable_licenses.clear()
			elif x[:1] == '-':
				acceptable_licenses.discard(x[1:])
			else:
				acceptable_licenses.add(x)

		license_str = lic
		if "?" in license_str:
			use = use.split()
		else:
			use = []

		license_struct = use_reduce(license_str, uselist=use, opconvert=True)
		return self._getMaskedLicenses(license_struct, acceptable_licenses)

	def _getMaskedLicenses(self, license_struct, acceptable_licenses):
		if not license_struct:
			return []
		if license_struct[0] == "||":
			ret = []
			for element in license_struct[1:]:
				if isinstance(element, list):
					if element:
						tmp = self._getMaskedLicenses(element, acceptable_licenses)
						if not tmp:
							return []
						ret.extend(tmp)
				else:
					if element in acceptable_licenses:
						return []
					ret.append(element)
			# Return all masked licenses, since we don't know which combination
			# (if any) the user will decide to unmask.
			return ret

		ret = []
		for element in license_struct:
			if isinstance(element, list):
				if element:
					ret.extend(self._getMaskedLicenses(element,
						acceptable_licenses))
			else:
				if element not in acceptable_licenses:
					ret.append(element)
		return ret

	def set_accept_license_str(self, accept_license_str):
		if accept_license_str != self._accept_license_str:
			self._accept_license_str = accept_license_str
			self._accept_license = tuple(self.expandLicenseTokens(accept_license_str.split()))