client: Move CLI to an own file

[pakfire.git] / src / pakfire / cli.py

#!/usr/bin/python3
###############################################################################
#                                                                             #
# Pakfire - The IPFire package management system                              #
# Copyright (C) 2011 Pakfire development team                                 #
#                                                                             #
# This program is free software: you can redistribute it and/or modify        #
# it under the terms of the GNU General Public License as published by        #
# the Free Software Foundation, either version 3 of the License, or           #
# (at your option) any later version.                                         #
#                                                                             #
# This program is distributed in the hope that it will be useful,             #
# but WITHOUT ANY WARRANTY; without even the implied warranty of              #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
# GNU General Public License for more details.                                #
#                                                                             #
# You should have received a copy of the GNU General Public License           #
# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

import argparse
import datetime
import logging
import os
import shutil
import signal
import sys

from . import _pakfire
from . import client
from . import daemon

from .constants import *
from .i18n import _

class Cli(object):
	default_path = "/"

	def parse_cli(self):
		parser = argparse.ArgumentParser(
			description = _("Pakfire command line interface"),
		)
		subparsers = parser.add_subparsers()

		# Add common arguments
		self._add_common_arguments(parser)

		parser.add_argument("--arch", "-a", nargs="?",
			help=_("Run pakfire for the given architecture"))
		parser.add_argument("--root", metavar="PATH", default="/",
			help=_("The path where pakfire should operate in"))

		# check
		check = subparsers.add_parser("check", help=_("Check the system for any errors"))
		check.set_defaults(func=self.handle_check)

		# clean
		clean = subparsers.add_parser("clean", help=_("Cleanup all temporary files"))
		clean.set_defaults(func=self.handle_clean)

		# execute
		execute = subparsers.add_parser("execute",
			help=_("Executes a command in the pakfire environment (useful for development)"))
		execute.add_argument("--bind", action="append", default=[], dest="binds",
			help=_("Bind-mounts the given directory"))
		execute.add_argument("command", nargs=argparse.REMAINDER)
		execute.set_defaults(func=self.handle_execute)

		# info
		info = subparsers.add_parser("info",
			help=_("Print some information about the given package(s)"))
		info.add_argument("--long", action="store_true",
			help=_("Show more information"))
		info.add_argument("package", help=_("Give at least the name of one package"))
		info.set_defaults(func=self.handle_info)

		# install
		install = subparsers.add_parser("install",
			help=_("Install one or more packages to the system"))
		install.add_argument("package", nargs="+",
			help=_("Give name of at least one package to install"))
		install.add_argument("--without-recommended", action="store_true",
			help=_("Don't install recommended packages"))
		install.add_argument("--allow-uninstall", action="store_true",
			help=_("Allow uninstalling packages"))
		install.add_argument("--allow-downgrade", action="store_true",
			help=_("Allow downgrading packages"))
		install.set_defaults(func=self.handle_install)

		# provides
		provides = subparsers.add_parser("provides",
			help=_("Get a list of packages that provide a given file or feature"))
		provides.add_argument("pattern", help=_("File or feature to search for"))
		provides.set_defaults(func=self.handle_provides)

		# requires
		requires = subparsers.add_parser("requires",
			help=_("Get a list of packages that require this dependency"))
		requires.add_argument("pattern", help=_("File or feature to search for"))
		requires.set_defaults(func=self.handle_requires)

		# reinstall
		reinstall = subparsers.add_parser("reinstall",
			help=_("Reinstall one or more packages"))
		reinstall.add_argument("package", nargs="+",
			help=_("Give name of at least one package to reinstall"))
		reinstall.set_defaults(func=self.handle_reinstall)

		# remove
		remove = subparsers.add_parser("remove",
			help=_("Remove one or more packages from the system"))
		remove.add_argument("package", nargs="+",
			help=_("Give name of at least one package to remove"))
		remove.add_argument("--keep-dependencies", action="store_true",
			help=_("Keep dependencies installed"))
		remove.set_defaults(func=self.handle_remove)

		# repolist
		repolist = subparsers.add_parser("repolist",
			help=_("List all currently enabled repositories"))
		repolist.set_defaults(func=self.handle_repolist)

		# search
		search = subparsers.add_parser("search", help=_("Search for a given pattern"))
		search.add_argument("pattern", help=_("A pattern to search for"))
		search.set_defaults(func=self.handle_search)

		# sync
		sync = subparsers.add_parser("sync",
			help=_("Sync all installed with the latest one in the distribution"))
		sync.add_argument("--keep-orphaned", action="store_true",
			help=_("Keep orphaned packages"))
		sync.set_defaults(func=self.handle_sync)

		# update
		update = subparsers.add_parser("update",
			help=_("Update the whole system or one specific package"))
		update.add_argument("package", nargs="*",
			help=_("Give a name of a package to update or leave emtpy for all"))
		update.add_argument("--exclude", "-x", nargs="+", default=[],
			help=_("Exclude package from update"))
		update.add_argument("--allow-uninstall", action="store_true",
			help=_("Allow uninstalling packages"))
		update.add_argument("--allow-downgrade", action="store_true",
			help=_("Allow downgrading packages"))
		update.set_defaults(func=self.handle_update)

		return parser.parse_args()

	def _add_common_arguments(self, parser, offline_switch=True):
		parser.add_argument("--version", action="version",
			version="%(prog)s " + PAKFIRE_VERSION)

		parser.add_argument("-v", "--verbose", action="store_true",
			help=_("Enable verbose output."))

		parser.add_argument("-c", "--config", nargs="?",
			help=_("Path to a configuration file to load."))

		parser.add_argument("--disable-repo", nargs="*", metavar="REPO",
			help=_("Disable a repository temporarily."), default=[])

		parser.add_argument("--enable-repo", nargs="*", metavar="REPO",
			help=_("Enable a repository temporarily."), default=[])

		if offline_switch:
			parser.add_argument("--offline", action="store_true",
				help=_("Run pakfire in offline mode."))

	def pakfire(self, ns):
		p = _pakfire.Pakfire(
			arch=ns.arch,
			conf=ns.config,
			path=ns.root if "root" in ns else self.default_path,
			offline=ns.offline,
		)

		# Disable repositories.
		for repo_name in ns.disable_repo:
			repo = p.get_repo(repo_name)
			repo.enabled = False

		# Enable repositories.
		for repo_name in ns.enable_repo:
			repo = p.get_repo(repo_name)
			repo.enabled = True

		return p

	def handle_execute(self, ns):
		pakfire = self.pakfire(ns)

		# Bind-mount everything
		for bind in ns.binds:
			pakfire.bind(bind)

		# Log everything to the console
		def logging_callback(level, line):
			if level >= logging.ERROR:
				sys.stderr.write("%s\n" % line)
			else:
				sys.stdout.write("%s\n" % line)

		return pakfire.execute(ns.command, logging_callback=logging_callback)

	def run(self):
		args = self.parse_cli()
		assert args.func, "Argument function not defined"

		try:
			return args.func(args)

		except KeyboardInterrupt:
			return 128 + signal.SIGINT

		except DependencyError as e:
			print(_("One or more dependencies could not been resolved"))
			print("") # empty line

			# This exception provides a list of all problems
			problems, = e.args

			# List all problems
			for problem in problems:
				print("  * %s" % problem)

				print("    %s" % _("Possible solutions are:"))
				for solution in problem.solutions:
					print("    * %s" % solution)

				# Add another empty line
				print("")

			return 4

		# Catch all errors and show a user-friendly error message.
		except Error as e:
			print(_("An error has occured when running Pakfire"))
			print(_("%s: %s") % (e.__class__.__name__, e.message))

			return e.exit_code

	def handle_info(self, ns):
		p = self.pakfire(ns)

		for pkg in p.search(ns.package, name_only=True):
			s = pkg.dump(long=ns.long)
			print(s)

	def handle_search(self, ns):
		p = self.pakfire(ns)

		for pkg in p.search(ns.pattern):
			# Skip any -debuginfo packages
			if pkg.name.endswith("-debuginfo"):
				continue

			print("%-24s: %s" % (pkg.name, pkg.summary))

	def handle_update(self, ns):
		p = self.pakfire(ns)
		p.update(
			ns.package,
			excludes=ns.exclude,
			allow_uninstall=ns.allow_uninstall,
			allow_downgrade=ns.allow_downgrade,
		)

	def handle_sync(self, ns):
		self.pakfire(ns).sync(keep_orphaned=ns.keep_orphaned)

	def handle_install(self, ns):
		p = self.pakfire(ns)
		p.install(
			ns.package,
			without_recommended=ns.without_recommended,
			allow_uninstall=ns.allow_uninstall,
			allow_downgrade=ns.allow_downgrade,
		)

	def handle_reinstall(self, ns):
		with self.pakfire(ns) as p:
			transaction = p.reinstall(ns.package)

			# Execute the transaction
			self._execute_transaction(transaction)

	def handle_remove(self, ns):
		p = self.pakfire(ns)
		p.erase(ns.package, keep_dependencies=ns.keep_dependencies)

	def handle_provides(self, ns, long=False):
		for pkg in self.pakfire(ns).whatprovides(ns.pattern):
			s = pkg.dump(long=long)
			print(s)

	def handle_requires(self, ns):
		for pkg in self.pakfire(ns).whatrequires(ns.pattern):
			s = pkg.dump(long=True)
			print(s)

	def handle_repolist(self, ns):
		p = self.pakfire(ns)

		FORMAT = " %-20s %8s %12s %12s "
		title = FORMAT % (_("Repository"), _("Enabled"), _("Priority"), _("Packages"))
		print(title)
		print("=" * len(title)) # spacing line

		for repo in p.repos:
			print(FORMAT % (repo.name, repo.enabled, repo.priority, len(repo)))

	def handle_clean(self, ns):
		print(_("Cleaning up everything..."))

		p = self.pakfire(ns)
		p.clean()

	def handle_check(self, ns):
		self.pakfire(ns).check()


class CliKey(Cli):
	def parse_cli(self):
		parser = argparse.ArgumentParser(
			description = _("Pakfire key command line interface"),
		)
		subparsers = parser.add_subparsers()

		# Add common arguments
		self._add_common_arguments(parser)

		# delete
		delete = subparsers.add_parser("delete", help=_("Delete a key from the local keyring"))
		delete.add_argument("fingerprint", nargs="+", help=_("The fingerprint of the key to delete"))
		delete.set_defaults(func=self.handle_delete)

		# export
		export = subparsers.add_parser("export", help=_("Export a key to a file"))
		export.add_argument("fingerprint", nargs=1,
			help=_("The fingerprint of the key to export"))
		export.add_argument("--filename", nargs="*", help=_("Write the key to this file"))
		export.add_argument("--secret", action="store_true",
			help=_("Export the secret key"))
		export.set_defaults(func=self.handle_export)

		# import
		_import = subparsers.add_parser("import", help=_("Import a key from file"))
		_import.add_argument("filename", nargs="+", help=_("Filename of that key to import"))
		_import.set_defaults(func=self.handle_import)

		# generate
		generate = subparsers.add_parser("generate", help=_("Import a key from file"))
		generate.add_argument("--realname", nargs=1,
			help=_("The real name of the owner of this key"))
		generate.add_argument("--email", nargs=1,
			help=_("The email address of the owner of this key"))
		generate.set_defaults(func=self.handle_generate)

		# list
		list = subparsers.add_parser("list", help=_("List all imported keys"))
		list.set_defaults(func=self.handle_list)

		return parser.parse_args()

	def parse_command_sign(self):
		# Implement the "sign" command.
		sub_sign = self.sub_commands.add_parser("sign",
			help=_("Sign one or more packages."))
		sub_sign.add_argument("--key", "-k", nargs=1,
			help=_("Key that is used sign the package(s)."))
		sub_sign.add_argument("package", nargs="+",
			help=_("Package(s) to sign."))
		sub_sign.add_argument("action", action="store_const", const="sign")

	def parse_command_verify(self):
		# Implement the "verify" command.
		sub_verify = self.sub_commands.add_parser("verify",
			help=_("Verify one or more packages."))
		#sub_verify.add_argument("--key", "-k", nargs=1,
		#	help=_("Key that is used verify the package(s)."))
		sub_verify.add_argument("package", nargs="+",
			help=_("Package(s) to verify."))
		sub_verify.add_argument("action", action="store_const", const="verify")

	def handle_generate(self, ns):
		p = self.pakfire(ns)

		print(_("Generating the key may take a moment..."))
		print()

		key = p.generate_key("%s <%s>" % (ns.realname.pop(), ns.email.pop()))

	def handle_list(self, ns):
		p = self.pakfire(ns)

		for key in p.keys:
			print(key)

	def handle_export(self, ns):
		p = self.pakfire(ns)

		for fingerprint in ns.fingerprint:
			key = p.get_key(fingerprint)
			if not key:
				print(_("Could not find key with fingerprint %s") % fingerprint)
				continue

			data = key.export(ns.secret)

			for file in ns.filename:
				with open(file, "w") as f:
					f.write(data)
			else:
				print(data)

	def handle_import(self, ns):
		p = self.pakfire(ns)

		for filename in ns.filename:
			with open(filename) as f:
				data = f.read()

				keys = p.import_key(data)
				for key in keys:
					print(key)

	def handle_delete(self, ns):
		p = self.pakfire(ns)

		for fingerprint in ns.fingerprint:
			key = p.get_key(fingerprint)
			if key:
				key.delete()

	def handle_sign(self):
		# Get the files from the command line options
		files = []

		for file in self.args.package:
			# Check, if we got a regular file
			if os.path.exists(file):
				file = os.path.abspath(file)
				files.append(file)

			else:
				raise FileNotFoundError(file)

		key = self.args.key[0]

		# Create pakfire instance.
		p = self.create_pakfire()

		for file in files:
			# Open the package.
			pkg = packages.open(p, None, file)

			print(_("Signing %s...") % pkg.friendly_name)
			pkg.sign(key)

	def handle_verify(self):
		# TODO needs to use new key code from libpakfire

		# Get the files from the command line options
		files = []

		for file in self.args.package:
			# Check, if we got a regular file
			if os.path.exists(file) and not os.path.isdir(file):
				file = os.path.abspath(file)
				files.append(file)

		# Create pakfire instance.
		p = self.create_pakfire()

		for file in files:
			# Open the package.
			pkg = packages.open(p, None, file)

			print(_("Verifying %s...") % pkg.friendly_name)
			sigs = pkg.verify()

			for sig in sigs:
				key = p.keyring.get_key(sig.fpr)
				if key:
					subkey = key.subkeys[0]

					print("  %s %s" % (subkey.fpr[-16:], key.uids[0].uid))
					if sig.validity:
						print("    %s" % _("This signature is valid."))

				else:
					print("  %s <%s>" % (sig.fpr, _("Unknown key")))
					print("    %s" % _("Could not check if this signature is valid."))

				created = datetime.datetime.fromtimestamp(sig.timestamp)
				print("    %s" % _("Created: %s") % created)

				if sig.exp_timestamp:
					expires = datetime.datetime.fromtimestamp(sig.exp_timestamp)
					print("    %s" % _("Expires: %s") % expires)

			print() # Empty line