pre-commit

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
#  pre-commit
#
#  Copyright 2016 Spencer McIntyre <zeroSteiner@gmail.com>
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are
#  met:
#
#  * Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
#  * Redistributions in binary form must reproduce the above
#    copyright notice, this list of conditions and the following disclaimer
#    in the documentation and/or other materials provided with the
#    distribution.
#  * Neither the name of the  nor the names of its
#    contributors may be used to endorse or promote products derived from
#    this software without specific prior written permission.
#
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
#  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
#  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
#  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
#  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
#  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
#  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
#  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
#  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
#  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#

# This file is used by the development team in order to ensure that the catalog
# is properly updated with a valid signature. This file should be symlinked to
# .git/hooks/pre-commit and assumes that King Phisher exists in a directory
# at the same level as the plugins, such that:
#
#   some/path/king-phisher
#   some/path/king-phisher-plugins
#
# The King Phisher development dependencies must have been installed with
# `pipenv install --dev`, and the operator should have a valid King Phisher
# signing key.
#
# Commands to merge a Pull Request:
#
#   1. git checkout master
#   2. git merge --edit --no-commit --no-ff pr/##
#   3. git commit
#
# The third command will run the pre-commit script and properly sign the
# necessary files. In Git v2.24 a new pre-merge-commit hook was made available
# which may streamline this process, but this has not been tested.

import argparse
import collections
import datetime
import getpass
import glob
import importlib
import os
import sys

import jinja2

__version__ = '1.0'

if sys.version_info < (3, 4):
	# python 3 is required to import without an __init__.py file
	print('this requires at least python version 3.4 to run')
	sys.exit(os.EX_SOFTWARE)

plugins_dir = os.path.dirname(os.path.abspath(__file__))
if plugins_dir.endswith(os.path.join('.git', 'hooks')):
	plugins_dir = os.path.normpath(os.path.join(plugins_dir, '..', '..'))
king_phisher_dir = os.path.normpath(os.path.join(plugins_dir, '..', 'king-phisher'))

sys.path.insert(0, king_phisher_dir)
sys.path.insert(0, plugins_dir)

try:
	import git
except ImportError:
	print('failed to import GitPython')
	sys.exit(os.EX_UNAVAILABLE)

try:
	import king_phisher.catalog as catalog
	import king_phisher.find as find
	import king_phisher.security_keys as security_keys
	import king_phisher.serializers as serializers
except ImportError:
	print('failed to import king_phisher')
	print('path assumed to be at: ' + king_phisher_dir)
	sys.exit(os.EX_UNAVAILABLE)

find.init_data_path()

PLUGIN_TYPES = ('client', 'server')
Plugin = collections.namedtuple('Plugin', ('class_', 'name', 'removed', 'type'))

def load_plugins(plugin_type, plugins_dir, verbose=False):
	plugins = []
	plugins_dir = os.path.join(plugins_dir, plugin_type)
	for plugin in glob.glob(os.path.join(plugins_dir, '*.py')):
		plugin = os.path.basename(plugin)
		plugin = os.path.splitext(plugin)[0]
		if verbose:
			print('loading ' + plugin_type + ' plugin: ' + plugin)
		plugin_module = importlib.import_module(plugin_type + '.' + plugin)
		plugins.append(plugin_module.Plugin)

	for plugin in os.listdir(plugins_dir):
		if not os.path.isdir(os.path.join(plugins_dir, plugin)):
			continue
		if not os.path.isfile(os.path.join(plugins_dir, plugin, '__init__.py')):
			continue
		if verbose:
			print('loading ' + plugin_type + ' plugin: ' + plugin)
		plugin_module = importlib.import_module(plugin_type + '.' + plugin)
		plugins.append(plugin_module.Plugin)
	plugins = sorted(plugins, key=lambda plugin: plugin.name)
	return plugins

def _make_plugin_collection(plugins_repo, plugin, signing_key, verbose=False):
	plugin_path = os.path.join(plugins_repo.working_tree_dir, plugin.type, plugin.name)
	if os.path.isfile(plugin_path + '.py'):
		plugin_path += '.py'
	elif not (os.path.isdir(plugin_path) and os.path.isfile(os.path.join(plugin_path, '__init__.py'))):
		raise RuntimeError("unknown path for {0} plugin: {1}".format(plugin.type, plugin.name))
	plugin_metadata = plugin.class_.metadata
	plugin_metadata.pop('is_compatible', None)

	if verbose:
		print('signing files for ' + plugin.type + ' plugin: ' + plugin.name)
	item_files = catalog.sign_item_files(plugin_path, signing_key, repo_path=plugins_dir)
	plugin_metadata['files'] = list(sorted(item_files, key=lambda item_file: item_file.path_source))
	for idx, item_file in enumerate(plugin_metadata['files']):
		plugin_metadata['files'][idx] = {
			'path-destination': item_file.path_destination,
			'path-source': item_file.path_source,
			'signature': item_file.signature,
			'signed-by': item_file.signed_by
		}
	return plugin_metadata

def _plugin_exists(plugins_repo, plugin_type, plugin_name):
	plugin_path = os.path.join(plugins_repo.working_tree_dir, plugin_type, plugin_name)
	if os.path.isfile(plugin_path + '.py'):
		return True
	elif os.path.isdir(plugin_path) and os.path.isfile(os.path.join(plugin_path, '__init__.py')):
		return True
	return False

def _get_all_plugins(plugins):
	new_plugins = collections.deque()
	for plugin_type in PLUGIN_TYPES:
		for plugin_class in plugins.get(plugin_type, []):
			new_plugins.append(Plugin(
				class_=plugin_class,
				name=plugin_class.name,
				removed=False,
				type=plugin_type
			))
	return new_plugins

def _get_modified_plugins(plugins, plugins_repo):
	path_to_name = lambda path: path.split(os.sep)[1]
	staged_plugin_files = collections.defaultdict(set)
	for diff in plugins_repo.index.diff('HEAD'):
		for staged_file in (diff.a_path, diff.b_path):
			plugin_type = staged_file.split(os.sep, 1)[0]
			if plugin_type not in PLUGIN_TYPES:
				continue
			plugin_name = path_to_name(staged_file)
			if plugin_name.endswith('.py'):
				plugin_name = plugin_name[:-3]
			staged_plugin_files[(plugin_type, plugin_name)].add(staged_file)

	new_plugins = collections.deque()
	for (plugin_type, plugin_name) in staged_plugin_files.keys():
		plugin_class = next((plugin_class for plugin_class in plugins[plugin_type] if plugin_class.name == plugin_name), None)
		new_plugins.append(Plugin(
			class_=plugin_class,
			name=plugin_name,
			removed=not _plugin_exists(plugins_repo, plugin_type, plugin_name),
			type=plugin_type
		))
	return new_plugins

def update_plugin_collections(collection, plugins_repo, plugins, signing_key, sign_all=True, verbose=False):
	# update the collection metadata
	collection['created'] = datetime.datetime.utcnow().isoformat() + '+00:00'
	collection['created-by'] = signing_key.id
	collection['signed-by'] = signing_key.id

	# get the plugins to process, either all or just the modified ones
	if sign_all:
		plugins = _get_all_plugins(plugins)
		# remove all existing collections
		collection['collections'] = {}
	else:
		plugins = _get_modified_plugins(plugins, plugins_repo)
		# preserve existing collections
		collection['collections'] = collection.get('collections', {})

	for plugin in plugins:
		plugin_collection = collection['collections'].get('plugins/' + plugin.type, [])
		# check if the plugin already exists...
		existing = next((plugin_metadata for plugin_metadata in plugin_collection if plugin_metadata['name'] == plugin.name), None)
		if existing:
			# and if so, remove the entry
			plugin_collection.remove(existing)
		if plugin.removed:
			continue
		plugin_collection.append(_make_plugin_collection(plugins_repo, plugin, signing_key, verbose=verbose))
		collection['collections']['plugins/' + plugin.type] = plugin_collection

	for plugin_type in set((plugin.type for plugin in plugins)):
		collection_type = 'plugins/' + plugin_type
		plugin_collection = collection['collections'].get(collection_type)
		if plugin_collection is None:
			continue
		plugin_collection = sorted(plugin_collection, key=lambda plugin: plugin['name'])
		collection['collections'][collection_type] = plugin_collection

def main():
	parser = argparse.ArgumentParser(description='pre-commit hook', conflict_handler='resolve')
	parser.add_argument('-k', '--key', default=os.getenv('KING_PHISHER_DEV_KEY'), help='path to a signing key')
	parser.add_argument('-V', '--verbose', action='store_true', default=False, help='print verbose output')
	parser.add_argument('-v', '--version', action='version', version='%(prog)s Version: ' + __version__)
	parser.add_argument('--git-no-add', action='store_false', default=True, dest='git_add', help='don\'t stage changed files in git')
	parser.add_argument('--sign-all', action='store_true', default=False, help='sign all files')
	arguments = parser.parse_args()

	if arguments.key is None:
		print('must specify a dev key path')
		return os.EX_USAGE
	if not os.path.isfile(arguments.key):
		print('invalid path to dev key: ' + arguments.key)
		return os.EX_NOINPUT
	if not os.access(arguments.key, os.R_OK):
		print('missing read privileges on dev key: ' + arguments.key)
		return os.EX_NOPERM

	try:
		signing_key = security_keys.SigningKey.from_file(
			arguments.key,
			password=(getpass.getpass('password: ') if arguments.key.endswith('.enc') else None)
		)
	except KeyboardInterrupt:
		return os.EX_TEMPFAIL
	except ValueError:
		print('failed to load the signing key')
		return os.EX_TEMPFAIL

	if arguments.verbose:
		print('plugins directory: ' + plugins_dir)
	plugins_repo = git.Repo(plugins_dir)

	jinja_env = jinja2.Environment(trim_blocks=True)
	jinja_env.filters['strftime'] = lambda dt, fmt: dt.strftime(fmt)
	with open(os.path.join(plugins_dir, 'README.jnj'), 'r') as file_h:
		readme_template = jinja_env.from_string(file_h.read())

	plugins = {
		'client': load_plugins('client', plugins_dir, verbose=arguments.verbose),
		'server': load_plugins('server', plugins_dir, verbose=arguments.verbose)
	}
	readme = readme_template.render(plugins=plugins, timestamp=datetime.datetime.utcnow())
	with open(os.path.join(plugins_dir, 'README.md'), 'w') as file_h:
		file_h.write(readme)
	if arguments.git_add:
		plugins_repo.index.add(['README.md'])

	with open(os.path.join(plugins_dir, 'catalog-collections.json'), 'r') as file_h:
		collection = serializers.JSON.load(file_h)

	update_plugin_collections(
		collection,
		plugins_repo,
		plugins,
		signing_key,
		sign_all=arguments.sign_all,
		verbose=arguments.verbose
	)

	collection = signing_key.sign_dict(collection)
	with open(os.path.join(plugins_dir, 'catalog-collections.json'), 'w') as file_h:
		serializers.JSON.dump(collection, file_h)
	if arguments.git_add:
		plugins_repo.index.add(['catalog-collections.json'])
	return os.EX_OK

if __name__ == '__main__':
	sys.exit(main())