Merge lp:~abentley/juju-core/sign-metadata into lp:~juju-qa/juju-core/cd-release-juju

[Aaron Bentley]

Add sign_metadata script.

This branch copies the sign_metadata script from juju-release-tools to cd-release-juju.

fake_gpg and its associated functions were extracted so that they could be reused.

[Curtis Hovey (sinzui)]

Thank you.

review: Approve (code)

=== added file 'Makefile'
--- Makefile	1970-01-01 00:00:00 +0000
+++ Makefile	2016-02-02 19:14:21 +0000
@@ -0,0 +1,11 @@
+p=*.py
+test:
+	TMPDIR=/tmp python -m unittest discover -vv ./tests -p "$(p)"
+lint:
+	flake8 --max-line-length=80 $$(find . -name '*.py')
+cover:
+	python -m coverage run --source="./" --omit "./tests/*" -m unittest discover -vv ./tests
+	python -m coverage report
+clean:
+	find . -name '*.pyc' -delete
+.PHONY: lint test cover clean

=== added file 'sign_metadata.py'
--- sign_metadata.py	1970-01-01 00:00:00 +0000
+++ sign_metadata.py	2016-02-02 19:14:21 +0000
@@ -0,0 +1,94 @@
+from __future__ import print_function
+
+from argparse import ArgumentParser
+from os import listdir
+from os.path import (
+    basename,
+    join,
+    isdir,
+    isfile,
+)
+import subprocess
+from tempfile import NamedTemporaryFile
+
+
+def run(command, verbose=False, dry_run=False):
+    """Run command list and ensure stdout and error are available."""
+    if verbose:
+        print(command)
+    try:
+        subprocess.check_output(command, stderr=subprocess.STDOUT)
+    except subprocess.CalledProcessError as e:
+        print('FAIL: {} - Returncode: {}'.format(e.output, e.returncode))
+        raise
+
+
+def sign_metadata(signing_key, meta_dir, signing_passphrase_file=None):
+    key_option, gpg_options = get_gpg_options(
+        signing_key, signing_passphrase_file)
+    for meta_file in get_meta_files(meta_dir):
+        with NamedTemporaryFile() as temp_file:
+            meta_file_path = join(meta_dir, meta_file)
+            update_file_content(meta_file_path, temp_file.name)
+            meta_basename = basename(meta_file).replace('.json', '.sjson')
+            output_file = join(meta_dir, meta_basename)
+            ensure_no_file(output_file)
+            cmd = 'gpg {} --clearsign {} -o {} {}'.format(
+                gpg_options, key_option, output_file, temp_file.name)
+            run(cmd.split())
+            output_file = join(meta_dir, '{}.gpg'.format(meta_file))
+            ensure_no_file(output_file)
+            cmd = 'gpg {} --detach-sign {} -o {} {}'.format(
+                gpg_options, key_option, output_file, meta_file_path)
+            run(cmd.split())
+
+
+def update_file_content(mete_file, dst_file):
+    with open(mete_file) as m:
+        with open(dst_file, 'w') as d:
+            d.write(m.read().replace('.json', '.sjson'))
+
+
+def get_gpg_options(signing_key, signing_passphrase_file=None):
+    key_option = "--default-key {}".format(signing_key)
+    gpg_options = "--no-tty"
+    if signing_passphrase_file:
+        gpg_options = "--no-use-agent --no-tty --passphrase-file {}".format(
+            signing_passphrase_file)
+    return key_option, gpg_options
+
+
+def get_meta_files(meta_dir):
+    meta_files = [f for f in listdir(meta_dir) if f.endswith('.json')]
+    if not meta_files:
+        print('Warning! no meta files found in {}'.format(meta_dir))
+    return meta_files
+
+
+def ensure_no_file(file_path):
+    if isfile(file_path):
+        raise ValueError('FAIL! file already exists: {}'.format(file_path))
+
+
+def parse_args(argv=None):
+    parser = ArgumentParser("Sign streams' meta files.")
+    parser.add_argument('metadata_dir', help='Metadata directory.')
+    parser.add_argument('signing_key', help='Key to sign with.')
+    parser.add_argument(
+        '-p', '--signing-passphrase-file', help='Signing passphrase file path.')
+    args = parser.parse_args(argv)
+    if not isdir(args.metadata_dir):
+        parser.error(
+            'Invalid metadata directory path {}'.format(args.metadata_dir))
+    if (args.signing_passphrase_file and
+            not isfile(args.signing_passphrase_file)):
+        parser.error(
+            'Invalid passphrase file path {}'.format(
+                args.signing_passphrase_file))
+    return args
+
+
+if __name__ == '__main__':
+    args = parse_args()
+    sign_metadata(args.signing_key, args.metadata_dir,
+                  args.signing_passphrase_file)

=== added directory 'tests'
=== added file 'tests/__init__.py'
--- tests/__init__.py	1970-01-01 00:00:00 +0000
+++ tests/__init__.py	2016-02-02 19:14:21 +0000
@@ -0,0 +1,14 @@
+from StringIO import StringIO
+from unittest import TestCase
+
+from mock import patch
+
+
+class QuietTestCase(TestCase):
+
+    def setUp(self):
+        super(QuietTestCase, self).setUp()
+        self.stdout = StringIO()
+        patcher = patch('sys.stdout', self.stdout)
+        patcher.start()
+        self.addCleanup(patcher.stop)

=== added file 'tests/test_sign_metadata.py'
--- tests/test_sign_metadata.py	1970-01-01 00:00:00 +0000
+++ tests/test_sign_metadata.py	2016-02-02 19:14:21 +0000
@@ -0,0 +1,200 @@
+from argparse import Namespace
+from contextlib import contextmanager
+from os.path import(
+    join,
+    isfile,
+)
+import json
+from StringIO import StringIO
+from tempfile import (
+    NamedTemporaryFile,
+    )
+from unittest import TestCase
+
+from mock import (
+    patch,
+    call,
+)
+
+from sign_metadata import (
+    get_gpg_options,
+    get_meta_files,
+    sign_metadata,
+    parse_args,
+    update_file_content,
+)
+from utility import temp_dir
+
+
+__metaclass__ = type
+
+
+def write_file(path, contents):
+    with open(path, 'w') as f:
+        f.write(contents)
+
+
+def gpg_header():
+    return '-----BEGIN PGP SIGNED MESSAGE-----\nHash: SHA1\n'
+
+
+def gpg_footer():
+    return ('\n-----BEGIN PGP SIGNATURE-----\nblah blah\n'
+            '-----END PGP SIGNATURE-----')
+
+
+def fake_gpg(args):
+    output_file = args[6]
+    input_file = args[7]
+    if '--passphrase-file' in args:
+        output_file = args[9]
+        input_file = args[10]
+    if '--clearsign' in args:
+        with open(input_file) as in_file:
+            with open(output_file, 'w') as out_file:
+                content = in_file.read()
+                out_file.write('{}{}{}'.format(
+                    gpg_header(), content, gpg_footer()))
+    if '--detach-sign' in args:
+        open(output_file, 'w').close()
+
+
+class TestSignMetaData(TestCase):
+
+    content = json.dumps(
+        {'index':
+            {'com.ubuntu.juju:released:tools':
+                {'path': "streams/v1/com.ubuntu.juju-released-tools.json",
+                 'format': 'products:1.0'}}
+         })
+    expected = json.dumps(
+        {'index':
+            {'com.ubuntu.juju:released:tools':
+                {'path': "streams/v1/com.ubuntu.juju-released-tools.sjson",
+                 "format": "products:1.0"}}
+         })
+
+    def test_get_meta_files(self):
+        with temp_dir() as meta_dir:
+            file1 = join(meta_dir, "index.json")
+            file2 = join(meta_dir, "proposed-tools.json")
+            file3 = join(meta_dir, "random.txt")
+            open(file1, 'w').close()
+            open(file2, 'w').close()
+            open(file3, 'w').close()
+            meta_files = get_meta_files(meta_dir)
+            self.assertItemsEqual(
+                meta_files, ['index.json', 'proposed-tools.json'])
+
+    def test_gpg_options(self):
+        output = get_gpg_options('[e-mail]')
+        expected = ("--default-key [e-mail]", '--no-tty')
+        self.assertEqual(output, expected)
+
+    def test_gpg_options_signing_passphrase_file(self):
+        output = get_gpg_options('[e-mail]', '/tmp')
+        expected = ("--default-key [e-mail]",
+                    '--no-use-agent --no-tty --passphrase-file /tmp')
+        self.assertEqual(output, expected)
+
+    def test_update_file_content(self):
+        with NamedTemporaryFile() as meta_file:
+            with NamedTemporaryFile() as dst_file:
+                write_file(meta_file.name, self.content)
+                update_file_content(meta_file.name, dst_file.name)
+                with open(dst_file.name) as dst:
+                    dst_file_content = dst.read()
+                self.assertEqual(dst_file_content, self.expected)
+
+    def test_sign_metadata(self):
+        with patch('sign_metadata.run', autospec=True,
+                   side_effect=fake_gpg) as smr:
+            with temp_dir() as meta_dir:
+                meta_file = join(meta_dir, 'index.json')
+                write_file(meta_file, self.content)
+                with NamedTemporaryFile() as temp_file:
+                    with patch('sign_metadata.NamedTemporaryFile',
+                               autospec=True, return_value=temp_file) as ntf:
+                        sign_metadata('[e-mail]', meta_dir)
+                        self.verify_signed_content(meta_dir)
+        signed_file = meta_file.replace('.json', '.sjson')
+        gpg_file = '{}.gpg'.format(meta_file)
+        calls = [
+            call(['gpg', '--no-tty', '--clearsign', '--default-key',
+                  '[e-mail]', '-o', signed_file, temp_file.name]),
+            call(['gpg', '--no-tty', '--detach-sign', '--default-key',
+                  '[e-mail]', '-o', gpg_file, meta_file])]
+        self.assertEqual(smr.mock_calls, calls)
+        ntf.assert_called_once_with()
+
+    def test_sign_metadata_signing_passphrase_file(self):
+        with patch('sign_metadata.run', autospec=True,
+                   side_effect=fake_gpg) as smr:
+            with temp_dir() as meta_dir:
+                meta_file = join(meta_dir, 'index.json')
+                write_file(meta_file, self.content)
+                with NamedTemporaryFile() as temp_file:
+                    with patch('sign_metadata.NamedTemporaryFile',
+                               autospec=True, return_value=temp_file) as ntf:
+                        sign_metadata(
+                            '[e-mail]', meta_dir, 'passphrase_file')
+                        self.verify_signed_content(meta_dir)
+        signed_file = meta_file.replace('.json', '.sjson')
+        gpg_file = '{}.gpg'.format(meta_file)
+        calls = [
+            call(['gpg', '--no-use-agent', '--no-tty', '--passphrase-file',
+                  'passphrase_file', '--clearsign', '--default-key',
+                  '[e-mail]', '-o', signed_file, temp_file.name]),
+            call(['gpg', '--no-use-agent', '--no-tty', '--passphrase-file',
+                  'passphrase_file', '--detach-sign', '--default-key',
+                  '[e-mail]', '-o', gpg_file, meta_file])]
+        self.assertEqual(smr.mock_calls, calls)
+        ntf.assert_called_once_with()
+
+    def verify_signed_content(self, meta_dir):
+        file_path = join(meta_dir, 'index.sjson')
+        with open(file_path) as i:
+            signed_content = i.read()
+        self.assertEqual(signed_content, '{}{}{}'.format(
+            gpg_header(), self.expected, gpg_footer()))
+        self.assertTrue(
+            isfile(join(meta_dir, 'index.json.gpg')))
+
+    def test_parse_args_default(self):
+        with temp_dir() as metadata_dir:
+            args = parse_args([metadata_dir, 's_key'])
+        self.assertEqual(args, Namespace(
+            signing_key='s_key', signing_passphrase_file=None,
+            metadata_dir=metadata_dir))
+
+    def test_parse_args_signing_passphrase_file(self):
+        with temp_dir() as metadata_dir:
+            with NamedTemporaryFile() as pass_file:
+                args = parse_args([metadata_dir, 's_key',
+                                   '--signing-passphrase-file', pass_file.name])
+        self.assertEqual(args, Namespace(
+            signing_key='s_key', signing_passphrase_file=pass_file.name,
+            metadata_dir=metadata_dir))
+
+    def test_parse_args_error(self):
+        with parse_error(self) as stderr:
+            parse_args([])
+        self.assertIn("error: too few arguments", stderr.getvalue())
+
+        with parse_error(self) as stderr:
+            parse_args(['metadata_dir', 'signing_key'])
+        self.assertIn("Invalid metadata directory path", stderr.getvalue())
+
+        with parse_error(self) as stderr:
+            with temp_dir() as metadata_dir:
+                parse_args([metadata_dir, 's_key', '--signing-passphrase-file',
+                            'fake/file/'])
+        self.assertIn("Invalid passphrase file path ", stderr.getvalue())
+
+
+@contextmanager
+def parse_error(test_case):
+    stderr = StringIO()
+    with test_case.assertRaises(SystemExit):
+        with patch('sys.stderr', stderr):
+            yield stderr

=== added file 'utility.py'
--- utility.py	1970-01-01 00:00:00 +0000
+++ utility.py	2016-02-02 19:14:21 +0000
@@ -0,0 +1,12 @@
+from contextlib import contextmanager
+import shutil
+from tempfile import mkdtemp
+
+
+@contextmanager
+def temp_dir():
+    dirname = mkdtemp()
+    try:
+        yield dirname
+    finally:
+        shutil.rmtree(dirname)