launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #22111
[Merge] lp:~cjwatson/launchpad/archive-signing-run-parts into lp:launchpad
Colin Watson has proposed merging lp:~cjwatson/launchpad/archive-signing-run-parts into lp:launchpad with lp:~cjwatson/launchpad/refactor-run-parts-subprocess as a prerequisite.
Commit message:
Extend IArchiveSigningKey to support signing files using run-parts.
Generic signing methods are now on ISignableArchive, with methods specific
to managed signing keys remaining on IArchiveSigningKey. This allows using
the same interface to sign files with either managed keys (the PPA case) or
external run-parts scripts (the case of the Ubuntu primary archive).
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/archive-signing-run-parts/+merge/336373
To keep this branch a manageable size, I haven't yet converted the call sites to take advantage of this; that will come in the next branch in this series.
https://code.launchpad.net/~cjwatson/ubuntu-archive-publishing/sign-parts/+merge/336347 will need to be deployed before this.
--
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~cjwatson/launchpad/archive-signing-run-parts into lp:launchpad.
=== modified file 'lib/lp/archivepublisher/archivesigningkey.py'
--- lib/lp/archivepublisher/archivesigningkey.py 2017-04-29 15:24:32 +0000
+++ lib/lp/archivepublisher/archivesigningkey.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2017 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""ArchiveSigningKey implementation."""
@@ -7,12 +7,18 @@
__all__ = [
'ArchiveSigningKey',
+ 'SignableArchive',
+ 'SigningMode',
]
import os
import gpgme
+from lazr.enum import (
+ EnumeratedType,
+ Item,
+ )
from twisted.internet.threads import deferToThread
from zope.component import getUtility
from zope.interface import implementer
@@ -24,7 +30,13 @@
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.archivepublisher.config import getPubConfig
from lp.archivepublisher.interfaces.archivesigningkey import (
+ CannotSignArchive,
IArchiveSigningKey,
+ ISignableArchive,
+ )
+from lp.archivepublisher.run_parts import (
+ find_run_parts_dir,
+ run_parts,
)
from lp.registry.interfaces.gpg import IGPGKeySet
from lp.services.config import config
@@ -32,13 +44,119 @@
from lp.services.propertycache import get_property_cache
+class SigningMode(EnumeratedType):
+ """Archive file signing mode."""
+
+ DETACHED = Item("Detached signature")
+ CLEAR = Item("Cleartext signature")
+
+
+@implementer(ISignableArchive)
+class SignableArchive:
+ """`IArchive` adapter for operations that involve signing files."""
+
+ gpgme_modes = {
+ SigningMode.DETACHED: gpgme.SIG_MODE_DETACH,
+ SigningMode.CLEAR: gpgme.SIG_MODE_CLEAR,
+ }
+
+ def __init__(self, archive):
+ self.archive = archive
+
+ @property
+ def can_sign(self):
+ """See `ISignableArchive`."""
+ return (
+ self.archive.signing_key is not None or
+ find_run_parts_dir(
+ self.archive.distribution.name, "sign.d") is not None)
+
+ def _makeSignatures(self, signatures, log=None):
+ """Make a sequence of signatures.
+
+ This abstraction is useful in the case where we're using an
+ in-process `GPGHandler`, since it avoids having to import the secret
+ key more than once.
+
+ :param signatures: A sequence of (input path, output path,
+ `SigningMode`, suite) tuples.
+ :param log: An optional logger.
+ """
+ if not self.can_sign:
+ raise CannotSignArchive(
+ "No signing key available for %s" % self.archive.displayname)
+
+ if self.archive.signing_key is not None:
+ secret_key_path = self.getPathForSecretKey(
+ self.archive.signing_key)
+ with open(secret_key_path) as secret_key_file:
+ secret_key_export = secret_key_file.read()
+ gpghandler = getUtility(IGPGHandler)
+ secret_key = gpghandler.importSecretKey(secret_key_export)
+
+ for input_path, output_path, mode, suite in signatures:
+ if self.archive.signing_key is not None:
+ with open(input_path) as input_file:
+ input_content = input_file.read()
+ signature = gpghandler.signContent(
+ input_content, secret_key, mode=self.gpgme_modes[mode])
+ with open(output_path, "w") as output_file:
+ output_file.write(signature)
+ elif find_run_parts_dir(
+ self.archive.distribution.name, "sign.d") is not None:
+ env = {
+ "INPUT_PATH": input_path,
+ "OUTPUT_PATH": output_path,
+ "MODE": mode.name.lower(),
+ "DISTRIBUTION": self.archive.distribution.name,
+ "SUITE": suite,
+ }
+ run_parts(
+ self.archive.distribution.name, "sign.d",
+ log=log, env=env)
+ else:
+ raise AssertionError(
+ "No signing key available for %s" %
+ self.archive.displayname)
+
+ def signRepository(self, suite, log=None):
+ """See `ISignableArchive`."""
+ suite_path = os.path.join(self._archive_root_path, 'dists', suite)
+ release_file_path = os.path.join(suite_path, 'Release')
+ if not os.path.exists(release_file_path):
+ raise AssertionError(
+ "Release file doesn't exist in the repository: %s" %
+ release_file_path)
+
+ self._makeSignatures([
+ (release_file_path, os.path.join(suite_path, 'Release.gpg'),
+ SigningMode.DETACHED, suite),
+ (release_file_path, os.path.join(suite_path, 'InRelease'),
+ SigningMode.CLEAR, suite),
+ ], log=log)
+
+ def signFile(self, suite, path, log=None):
+ """See `ISignableArchive`."""
+ # Allow the passed path to be relative to the archive root.
+ path = os.path.realpath(os.path.join(self._archive_root_path, path))
+
+ # Ensure the resulting path is within the archive root after
+ # normalisation.
+ # NOTE: uses os.sep to prevent /var/tmp/../tmpFOO attacks.
+ archive_root = self._archive_root_path + os.sep
+ if not path.startswith(archive_root):
+ raise AssertionError(
+ "Attempting to sign file (%s) outside archive_root for %s" % (
+ path, self.archive.displayname))
+
+ self._makeSignatures(
+ [(path, "%s.gpg" % path, SigningMode.DETACHED, suite)], log=log)
+
+
@implementer(IArchiveSigningKey)
-class ArchiveSigningKey:
+class ArchiveSigningKey(SignableArchive):
"""`IArchive` adapter for manipulating its GPG key."""
- def __init__(self, archive):
- self.archive = archive
-
@property
def _archive_root_path(self):
return getPubConfig(self.archive).archiveroot
@@ -139,67 +257,3 @@
else:
pub_key = self._uploadPublicSigningKey(secret_key)
self._storeSigningKey(pub_key)
-
- def signRepository(self, suite):
- """See `IArchiveSigningKey`."""
- assert self.archive.signing_key is not None, (
- "No signing key available for %s" % self.archive.displayname)
-
- suite_path = os.path.join(self._archive_root_path, 'dists', suite)
- release_file_path = os.path.join(suite_path, 'Release')
- assert os.path.exists(release_file_path), (
- "Release file doesn't exist in the repository: %s"
- % release_file_path)
-
- secret_key_path = self.getPathForSecretKey(self.archive.signing_key)
- with open(secret_key_path) as secret_key_file:
- secret_key_export = secret_key_file.read()
-
- gpghandler = getUtility(IGPGHandler)
- secret_key = gpghandler.importSecretKey(secret_key_export)
-
- with open(release_file_path) as release_file:
- release_file_content = release_file.read()
- signature = gpghandler.signContent(
- release_file_content, secret_key, mode=gpgme.SIG_MODE_DETACH)
-
- release_signature_path = os.path.join(suite_path, 'Release.gpg')
- with open(release_signature_path, 'w') as release_signature_file:
- release_signature_file.write(signature)
-
- inline_release = gpghandler.signContent(
- release_file_content, secret_key, mode=gpgme.SIG_MODE_CLEAR)
-
- inline_release_path = os.path.join(suite_path, 'InRelease')
- with open(inline_release_path, 'w') as inline_release_file:
- inline_release_file.write(inline_release)
-
- def signFile(self, path):
- """See `IArchiveSigningKey`."""
- assert self.archive.signing_key is not None, (
- "No signing key available for %s" % self.archive.displayname)
-
- # Allow the passed path to be relative to the archive root.
- path = os.path.realpath(os.path.join(self._archive_root_path, path))
-
- # Ensure the resulting path is within the archive root after
- # normalisation.
- # NOTE: uses os.sep to prevent /var/tmp/../tmpFOO attacks.
- archive_root = self._archive_root_path + os.sep
- assert path.startswith(archive_root), (
- "Attempting to sign file (%s) outside archive_root for %s" % (
- path, self.archive.displayname))
-
- secret_key_path = self.getPathForSecretKey(self.archive.signing_key)
- with open(secret_key_path) as secret_key_file:
- secret_key_export = secret_key_file.read()
- gpghandler = getUtility(IGPGHandler)
- secret_key = gpghandler.importSecretKey(secret_key_export)
-
- with open(path) as path_file:
- file_content = path_file.read()
- signature = gpghandler.signContent(
- file_content, secret_key, mode=gpgme.SIG_MODE_DETACH)
-
- with open(os.path.join(path + '.gpg'), 'w') as signature_file:
- signature_file.write(signature)
=== modified file 'lib/lp/archivepublisher/customupload.py'
--- lib/lp/archivepublisher/customupload.py 2016-06-14 15:02:46 +0000
+++ lib/lp/archivepublisher/customupload.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2016 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Infrastructure for handling custom uploads.
@@ -128,7 +128,7 @@
self.setTargetDirectory(archive, tarfile_path, suite)
self.checkForConflicts()
self.extract()
- self.installFiles()
+ self.installFiles(archive, suite)
self.fixCurrentSymlink()
finally:
self.cleanup()
@@ -268,7 +268,7 @@
if not os.path.isdir(parentdir):
os.makedirs(parentdir, 0o755)
- def installFiles(self):
+ def installFiles(self, archive, suite):
"""Install the files from the custom upload to the archive."""
assert self.tmpdir is not None, "Must extract tarfile first"
extracted = False
=== modified file 'lib/lp/archivepublisher/interfaces/archivesigningkey.py'
--- lib/lp/archivepublisher/interfaces/archivesigningkey.py 2017-04-29 15:24:32 +0000
+++ lib/lp/archivepublisher/interfaces/archivesigningkey.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2017 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""ArchiveSigningKey interface."""
@@ -6,17 +6,61 @@
__metaclass__ = type
__all__ = [
+ 'CannotSignArchive',
'IArchiveSigningKey',
+ 'ISignableArchive',
]
-from zope.interface import Interface
+from zope.interface import (
+ Attribute,
+ Interface,
+ )
from zope.schema import Object
from lp import _
from lp.soyuz.interfaces.archive import IArchive
-class IArchiveSigningKey(Interface):
+class CannotSignArchive(Exception):
+ """An archive is not set up for signing."""
+
+
+class ISignableArchive(Interface):
+ """`SignableArchive` interface.
+
+ `IArchive` adapter for operations that involve signing files.
+ """
+
+ archive = Object(
+ title=_('Corresponding IArchive'), required=True, schema=IArchive)
+
+ can_sign = Attribute("True if this archive is set up for signing.")
+
+ def signRepository(suite, log=None):
+ """Sign the corresponding repository.
+
+ :param suite: suite name to be signed.
+ :param log: an optional logger.
+ :raises CannotSignArchive: if the context archive is not set up for
+ signing.
+ :raises AssertionError: if there is no Release file in the given
+ suite.
+ """
+
+ def signFile(suite, path, log=None):
+ """Sign the corresponding file.
+
+ :param suite: name of the suite containing the file to be signed.
+ :param path: path within dists to sign with the archive key.
+ :param log: an optional logger.
+ :raises CannotSignArchive: if the context archive is not set up for
+ signing.
+ :raises AssertionError: if the given 'path' is outside of the
+ archive root.
+ """
+
+
+class IArchiveSigningKey(ISignableArchive):
"""`ArchiveSigningKey` interface.
`IArchive` adapter for operations using its 'signing_key'.
@@ -25,9 +69,6 @@
new signing keys.
"""
- archive = Object(
- title=_('Corresponding IArchive'), required=True, schema=IArchive)
-
def getPathForSecretKey(key):
"""Return the absolute path to access a secret key export.
@@ -80,20 +121,3 @@
`signing_key`.
:raises AssertionError: if the given 'key_path' does not exist.
"""
-
- def signRepository(suite):
- """Sign the corresponding repository.
-
- :param suite: suite name to be signed.
- :raises AssertionError: if the context archive has no `signing_key`
- or there is no Release file in the given suite.
- """
-
- def signFile(path):
- """Sign the corresponding file.
-
- :param path: path within dists to sign with the archive key.
- :raises AssertionError: if the context archive has no `signing_key`.
- :raises AssertionError: if the given 'path' is outside of the
- archive root.
- """
=== modified file 'lib/lp/archivepublisher/publishing.py'
--- lib/lp/archivepublisher/publishing.py 2016-06-17 21:17:58 +0000
+++ lib/lp/archivepublisher/publishing.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2016 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
__all__ = [
@@ -1478,10 +1478,9 @@
class DirectoryHash:
"""Represents a directory hierarchy for hashing."""
- def __init__(self, root, tmpdir, signer=None):
+ def __init__(self, root, tmpdir):
self.root = root
self.tmpdir = tmpdir
- self.signer = signer
self.checksum_hash = []
for usable in self._usable_archive_hashes:
@@ -1501,6 +1500,11 @@
if archive_hash.write_directory_hash:
yield archive_hash
+ @property
+ def checksum_paths(self):
+ for checksum_path, _, _ in self.checksum_hash:
+ yield checksum_path
+
def add(self, path):
"""Add a path to be checksummed."""
hashes = [
@@ -1524,5 +1528,3 @@
def close(self):
for (checksum_path, checksum_file, archive_hash) in self.checksum_hash:
checksum_file.close()
- if self.signer:
- self.signer.signFile(checksum_path)
=== modified file 'lib/lp/archivepublisher/signing.py'
--- lib/lp/archivepublisher/signing.py 2017-07-24 16:48:47 +0000
+++ lib/lp/archivepublisher/signing.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2016 Canonical Ltd. This software is licensed under the
+# Copyright 2012-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""The processing of Signing tarballs.
@@ -368,19 +368,19 @@
if 'tarball' in self.signing_options:
self.convertToTarball()
- def installFiles(self):
+ def installFiles(self, archive, suite):
"""After installation hash and sign the installed result."""
# Avoid circular import.
from lp.archivepublisher.publishing import DirectoryHash
- super(SigningUpload, self).installFiles()
+ super(SigningUpload, self).installFiles(archive, suite)
versiondir = os.path.join(self.targetdir, self.version)
- signer = None
- if self.archive.signing_key:
- signer = IArchiveSigningKey(self.archive)
- with DirectoryHash(versiondir, self.temproot, signer) as hasher:
+ with DirectoryHash(versiondir, self.temproot) as hasher:
hasher.add_dir(versiondir)
+ for checksum_path in hasher.checksum_paths:
+ if archive.signing_key is not None:
+ IArchiveSigningKey(archive).signFile(suite, checksum_path)
def shouldInstall(self, filename):
return filename.startswith("%s/" % self.version)
=== modified file 'lib/lp/archivepublisher/tests/archive-signing.txt'
--- lib/lp/archivepublisher/tests/archive-signing.txt 2017-08-03 14:26:40 +0000
+++ lib/lp/archivepublisher/tests/archive-signing.txt 2018-01-19 16:32:05 +0000
@@ -453,7 +453,7 @@
>>> archive_signing_key.signRepository(test_suite)
Traceback (most recent call last):
...
- AssertionError: No signing key available for PPA for Celso Providelo
+ CannotSignArchive: No signing key available for PPA for Celso Providelo
We'll purge 'signing_keys_root' and the PPA repository root so that
other tests don't choke on it, and shut down the server.
=== modified file 'lib/lp/archivepublisher/tests/test_archivesigningkey.py'
--- lib/lp/archivepublisher/tests/test_archivesigningkey.py 2017-04-29 15:24:32 +0000
+++ lib/lp/archivepublisher/tests/test_archivesigningkey.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2016-2017 Canonical Ltd. This software is licensed under the
+# Copyright 2016-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Test ArchiveSigningKey."""
@@ -6,16 +6,20 @@
__metaclass__ = type
import os
+from textwrap import dedent
from testtools.deferredruntest import AsynchronousDeferredRunTest
+from testtools.matchers import FileContains
from twisted.internet import defer
from zope.component import getUtility
from lp.archivepublisher.config import getPubConfig
from lp.archivepublisher.interfaces.archivesigningkey import (
IArchiveSigningKey,
+ ISignableArchive,
)
from lp.archivepublisher.interfaces.publisherconfig import IPublisherConfigSet
+from lp.archivepublisher.tests.test_run_parts import RunPartsMixin
from lp.services.osutils import write_file
from lp.soyuz.enums import ArchivePurpose
from lp.testing import TestCaseWithFactory
@@ -24,14 +28,14 @@
from lp.testing.layers import ZopelessDatabaseLayer
-class TestArchiveSigningKey(TestCaseWithFactory):
+class TestSignableArchiveWithSigningKey(TestCaseWithFactory):
layer = ZopelessDatabaseLayer
run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=10)
@defer.inlineCallbacks
def setUp(self):
- super(TestArchiveSigningKey, self).setUp()
+ super(TestSignableArchiveWithSigningKey, self).setUp()
self.temp_dir = self.makeTemporaryDirectory()
self.distro = self.factory.makeDistribution()
db_pubconf = getUtility(IPublisherConfigSet).getByDistribution(
@@ -48,39 +52,106 @@
yield IArchiveSigningKey(self.archive).setSigningKey(
key_path, async_keyserver=True)
- def test_signfile_absolute_within_archive(self):
+ def test_signFile_absolute_within_archive(self):
filename = os.path.join(self.archive_root, "signme")
write_file(filename, "sign this")
- signer = IArchiveSigningKey(self.archive)
- signer.signFile(filename)
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
+ signer.signFile(self.suite, filename)
signature = filename + '.gpg'
self.assertTrue(os.path.exists(signature))
- def test_signfile_absolute_outside_archive(self):
+ def test_signFile_absolute_outside_archive(self):
filename = os.path.join(self.temp_dir, "signme")
write_file(filename, "sign this")
- signer = IArchiveSigningKey(self.archive)
- self.assertRaises(AssertionError, lambda: signer.signFile(filename))
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
+ self.assertRaises(
+ AssertionError, lambda: signer.signFile(self.suite, filename))
- def test_signfile_relative_within_archive(self):
+ def test_signFile_relative_within_archive(self):
filename_relative = "signme"
filename = os.path.join(self.archive_root, filename_relative)
write_file(filename, "sign this")
- signer = IArchiveSigningKey(self.archive)
- signer.signFile(filename_relative)
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
+ signer.signFile(self.suite, filename_relative)
signature = filename + '.gpg'
self.assertTrue(os.path.exists(signature))
- def test_signfile_relative_outside_archive(self):
+ def test_signFile_relative_outside_archive(self):
filename_relative = "../signme"
filename = os.path.join(self.temp_dir, filename_relative)
write_file(filename, "sign this")
- signer = IArchiveSigningKey(self.archive)
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
self.assertRaises(
- AssertionError, lambda: signer.signFile(filename_relative))
+ AssertionError,
+ lambda: signer.signFile(self.suite, filename_relative))
+
+
+class TestSignableArchiveWithRunParts(RunPartsMixin, TestCaseWithFactory):
+
+ layer = ZopelessDatabaseLayer
+
+ def setUp(self):
+ super(TestSignableArchiveWithRunParts, self).setUp()
+ self.temp_dir = self.makeTemporaryDirectory()
+ self.distro = self.factory.makeDistribution()
+ db_pubconf = getUtility(IPublisherConfigSet).getByDistribution(
+ self.distro)
+ db_pubconf.root_dir = unicode(self.temp_dir)
+ self.archive = self.factory.makeArchive(
+ distribution=self.distro, purpose=ArchivePurpose.PRIMARY)
+ self.archive_root = getPubConfig(self.archive).archiveroot
+ self.suite = "distroseries"
+ self.enableRunParts(distribution_name=self.distro.name)
+ with open(os.path.join(
+ self.parts_directory, self.distro.name, "sign.d",
+ "10-sign"), "w") as sign_script:
+ sign_script.write(dedent("""\
+ #! /bin/sh
+ echo "$MODE signature of $INPUT_PATH ($DISTRIBUTION/$SUITE)" \\
+ >"$OUTPUT_PATH"
+ """))
+ os.fchmod(sign_script.fileno(), 0o755)
+
+ def test_signRepository_runs_parts(self):
+ suite_dir = os.path.join(self.archive_root, "dists", self.suite)
+ release_path = os.path.join(suite_dir, "Release")
+ write_file(release_path, "Release contents")
+
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
+ signer.signRepository(self.suite)
+
+ self.assertThat(
+ os.path.join(suite_dir, "Release.gpg"),
+ FileContains(
+ "detached signature of %s (%s/%s)\n" %
+ (release_path, self.distro.name, self.suite)))
+ self.assertThat(
+ os.path.join(suite_dir, "InRelease"),
+ FileContains(
+ "clear signature of %s (%s/%s)\n" %
+ (release_path, self.distro.name, self.suite)))
+
+ def test_signFile_runs_parts(self):
+ filename = os.path.join(self.archive_root, "signme")
+ write_file(filename, "sign this")
+
+ signer = ISignableArchive(self.archive)
+ self.assertTrue(signer.can_sign)
+ signer.signFile(self.suite, filename)
+
+ self.assertThat(
+ "%s.gpg" % filename,
+ FileContains(
+ "detached signature of %s (%s/%s)\n" %
+ (filename, self.distro.name, self.suite)))
=== modified file 'lib/lp/archivepublisher/tests/test_publisher.py'
--- lib/lp/archivepublisher/tests/test_publisher.py 2017-04-29 15:24:32 +0000
+++ lib/lp/archivepublisher/tests/test_publisher.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2017 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for publisher class."""
@@ -68,7 +68,6 @@
Publisher,
)
from lp.archivepublisher.utils import RepositoryIndexFile
-from lp.archivepublisher.interfaces.publisherconfig import IPublisherConfigSet
from lp.registry.interfaces.distribution import IDistributionSet
from lp.registry.interfaces.distroseries import IDistroSeries
from lp.registry.interfaces.person import IPersonSet
@@ -3208,16 +3207,6 @@
result[dh_file].append(line.strip().split(' '))
return result
- def fetchSigs(self, rootdir):
- result = defaultdict(list)
- for dh_file in self.all_hash_files:
- checksum_sig = os.path.join(rootdir, dh_file) + '.gpg'
- if os.path.exists(checksum_sig):
- with open(checksum_sig, "r") as sfd:
- for line in sfd:
- result[dh_file].append(line)
- return result
-
class TestDirectoryHash(TestDirectoryHashHelpers):
"""Unit tests for DirectoryHash object."""
@@ -3232,7 +3221,7 @@
checksum_file = os.path.join(rootdir, dh_file)
self.assertFalse(os.path.exists(checksum_file))
- with DirectoryHash(rootdir, tmpdir, None):
+ with DirectoryHash(rootdir, tmpdir):
pass
for dh_file in self.all_hash_files:
@@ -3295,63 +3284,3 @@
),
}
self.assertThat(self.fetchSums(rootdir), MatchesDict(expected))
-
-
-class TestDirectoryHashSigning(TestDirectoryHashHelpers):
- """Unit tests for DirectoryHash object, signing functionality."""
-
- layer = ZopelessDatabaseLayer
- run_tests_with = AsynchronousDeferredRunTest.make_factory(timeout=10)
-
- @defer.inlineCallbacks
- def setUp(self):
- super(TestDirectoryHashSigning, self).setUp()
- self.temp_dir = self.makeTemporaryDirectory()
- self.distro = self.factory.makeDistribution()
- db_pubconf = getUtility(IPublisherConfigSet).getByDistribution(
- self.distro)
- db_pubconf.root_dir = unicode(self.temp_dir)
- self.archive = self.factory.makeArchive(
- distribution=self.distro, purpose=ArchivePurpose.PRIMARY)
- self.archive_root = getPubConfig(self.archive).archiveroot
- self.suite = "distroseries"
-
- # Setup a keyserver so we can install the archive key.
- with InProcessKeyServerFixture() as keyserver:
- yield keyserver.start()
- key_path = os.path.join(gpgkeysdir, 'ppa-sample@xxxxxxxxxxxxxxxxx')
- yield IArchiveSigningKey(self.archive).setSigningKey(
- key_path, async_keyserver=True)
-
- def test_basic_directory_add_signed(self):
- tmpdir = unicode(self.makeTemporaryDirectory())
- rootdir = self.archive_root
- os.makedirs(rootdir)
-
- test1_file = os.path.join(rootdir, "test1")
- test1_hash = self.createTestFile(test1_file, "test1 dir")
-
- test2_file = os.path.join(rootdir, "test2")
- test2_hash = self.createTestFile(test2_file, "test2 dir")
-
- os.mkdir(os.path.join(rootdir, "subdir1"))
-
- test3_file = os.path.join(rootdir, "subdir1", "test3")
- test3_hash = self.createTestFile(test3_file, "test3 dir")
-
- signer = IArchiveSigningKey(self.archive)
- with DirectoryHash(rootdir, tmpdir, signer=signer) as dh:
- dh.add_dir(rootdir)
-
- expected = {
- 'SHA256SUMS': MatchesSetwise(
- Equals([test1_hash, "*test1"]),
- Equals([test2_hash, "*test2"]),
- Equals([test3_hash, "*subdir1/test3"]),
- ),
- }
- self.assertThat(self.fetchSums(rootdir), MatchesDict(expected))
- sig_content = self.fetchSigs(rootdir)
- for dh_file in sig_content:
- self.assertEqual(
- sig_content[dh_file][0], '-----BEGIN PGP SIGNATURE-----\n')
=== modified file 'lib/lp/archivepublisher/tests/test_run_parts.py'
--- lib/lp/archivepublisher/tests/test_run_parts.py 2018-01-19 16:32:05 +0000
+++ lib/lp/archivepublisher/tests/test_run_parts.py 2018-01-19 16:32:05 +0000
@@ -30,17 +30,18 @@
class RunPartsMixin:
"""Helper for run-parts tests."""
- def enableRunParts(self, parts_directory=None):
+ def enableRunParts(self, parts_directory=None, distribution_name="ubuntu"):
"""Set up for run-parts execution.
:param parts_directory: Base location for the run-parts directories.
If omitted, a temporary directory will be used.
+ :param distribution_name: The name of the distribution to set up.
"""
if parts_directory is None:
parts_directory = self.makeTemporaryDirectory()
- os.makedirs(os.path.join(
- parts_directory, "ubuntu", "publish-distro.d"))
- os.makedirs(os.path.join(parts_directory, "ubuntu", "finalize.d"))
+ for name in ("sign.d", "publish-distro.d", "finalize.d"):
+ os.makedirs(os.path.join(
+ parts_directory, distribution_name, name))
self.parts_directory = parts_directory
self.pushConfig("archivepublisher", run_parts_location=parts_directory)
=== modified file 'lib/lp/archivepublisher/tests/test_signing.py'
--- lib/lp/archivepublisher/tests/test_signing.py 2017-08-02 19:13:48 +0000
+++ lib/lp/archivepublisher/tests/test_signing.py 2018-01-19 16:32:05 +0000
@@ -1,4 +1,4 @@
-# Copyright 2012-2017 Canonical Ltd. This software is licensed under the
+# Copyright 2012-2018 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Test UEFI custom uploads."""
@@ -13,10 +13,12 @@
from testtools.deferredruntest import AsynchronousDeferredRunTest
from testtools.matchers import (
Contains,
+ FileContains,
Matcher,
MatchesAll,
Mismatch,
Not,
+ StartsWith,
)
from twisted.internet import defer
from zope.component import getUtility
@@ -895,7 +897,10 @@
sha256file = os.path.join(self.getSignedPath("test", "amd64"),
"1.0", "SHA256SUMS")
self.assertTrue(os.path.exists(sha256file))
- self.assertTrue(os.path.exists(sha256file + '.gpg'))
+ self.assertThat(
+ sha256file + '.gpg',
+ FileContains(
+ matcher=StartsWith('-----BEGIN PGP SIGNATURE-----\n')))
@defer.inlineCallbacks
def test_checksumming_tree_signed_options_tarball(self):
@@ -915,7 +920,10 @@
sha256file = os.path.join(self.getSignedPath("test", "amd64"),
"1.0", "SHA256SUMS")
self.assertTrue(os.path.exists(sha256file))
- self.assertTrue(os.path.exists(sha256file + '.gpg'))
+ self.assertThat(
+ sha256file + '.gpg',
+ FileContains(
+ matcher=StartsWith('-----BEGIN PGP SIGNATURE-----\n')))
tarfilename = os.path.join(self.getSignedPath("test", "amd64"),
"1.0", "signed.tar.gz")
Follow ups
