Merge lp:~cjwatson/launchpad/remove-deb-ar-assumption into lp:launchpad

[Colin Watson]

Check the .deb format using dpkg-deb rather than ar.

Debian has been talking about changing the outer .deb container file format to something other than ar, e.g. in order to lift size limits. It may or may not go anywhere, but it reminded me that we really shouldn't be making any particular assumptions about what the container format is: the interface we should rely on is that provided by dpkg.

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'lib/lp/archiveuploader/nascentuploadfile.py'
--- lib/lp/archiveuploader/nascentuploadfile.py	2018-05-03 15:10:39 +0000
+++ lib/lp/archiveuploader/nascentuploadfile.py	2019-05-24 17:19:17 +0000
@@ -1,4 +1,4 @@
-# Copyright 2009-2018 Canonical Ltd.  This software is licensed under the
+# Copyright 2009-2019 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 """Specific models for uploaded files"""
@@ -697,41 +697,26 @@
     def verifyFormat(self):
         """Check if the DEB format is sane.
 
-        Debian packages are in fact 'ar' files. Thus we run '/usr/bin/ar'
-        to look at the contents of the deb files to confirm they make sense.
+        We run 'dpkg-deb' to look at the contents of the deb files to
+        confirm they make sense.
         """
-        ar_process = subprocess.Popen(
-            ["/usr/bin/ar", "t", self.filepath],
-            stdout=subprocess.PIPE)
-        output = ar_process.stdout.read()
-        result = ar_process.wait()
-        if result != 0:
-            yield UploadError(
-                "%s: 'ar t' invocation failed." % self.filename)
-            yield UploadError(
-                prefix_multi_line_string(output, " [ar output:] "))
-
-        chunks = output.strip().split("\n")
-        if len(chunks) != 3:
-            yield UploadError(
-                "%s: found %d chunks, expecting 3. %r" % (
-                self.filename, len(chunks), chunks))
-
-        debian_binary, control_tar, data_tar = chunks
-        if debian_binary != "debian-binary":
-            yield UploadError(
-                "%s: first chunk is %s, expected debian-binary." % (
-                self.filename, debian_binary))
-        if control_tar not in ("control.tar.gz", "control.tar.xz"):
-            yield UploadError(
-                "%s: second chunk is %s, expected control.tar.gz or "
-                "control.tar.xz." % (self.filename, control_tar))
-        if data_tar not in ("data.tar.gz", "data.tar.bz2", "data.tar.lzma",
-                            "data.tar.xz"):
-            yield UploadError(
-                "%s: third chunk is %s, expected data.tar.gz, "
-                "data.tar.bz2, data.tar.lzma or data.tar.xz." %
-                (self.filename, data_tar))
+        try:
+            subprocess.check_output(
+                ["dpkg-deb", "-I", self.filepath], stderr=subprocess.STDOUT)
+        except subprocess.CalledProcessError as e:
+            yield UploadError(
+                "%s: 'dpkg-deb -I' invocation failed." % self.filename)
+            yield UploadError(
+                prefix_multi_line_string(e.output, " [dpkg-deb output:] "))
+
+        try:
+            subprocess.check_output(
+                ["dpkg-deb", "-c", self.filepath], stderr=subprocess.STDOUT)
+        except subprocess.CalledProcessError as e:
+            yield UploadError(
+                "%s: 'dpkg-deb -c' invocation failed." % self.filename)
+            yield UploadError(
+                prefix_multi_line_string(e.output, " [dpkg-deb output:] "))
 
     def verifyDebTimestamp(self):
         """Check specific DEB format timestamp checks."""

=== modified file 'lib/lp/archiveuploader/tests/test_nascentuploadfile.py'
--- lib/lp/archiveuploader/tests/test_nascentuploadfile.py	2019-05-24 17:19:17 +0000
+++ lib/lp/archiveuploader/tests/test_nascentuploadfile.py	2019-05-24 17:19:17 +0000
@@ -7,19 +7,29 @@
 
 __metaclass__ = type
 
+from functools import partial
+import gzip
 import hashlib
+import io
 import os
 import subprocess
+import tarfile
 
 from debian.deb822 import (
     Changes,
+    Deb822,
     Dsc,
     )
+try:
+    import lzma
+except ImportError:
+    from backports import lzma
 from testtools.matchers import (
     Contains,
     Equals,
     MatchesAny,
     MatchesListwise,
+    MatchesRegex,
     )
 
 from lp.archiveuploader.changesfile import ChangesFile
@@ -360,11 +370,23 @@
             "Priority": b"optional",
             "Homepage": b"http://samba.org/~jelmer/dulwich",
             "Description": b"Pure-python Git library\n"
-                b"Dulwich is a Python implementation of the file formats and "
-                b"protocols",
+                b" Dulwich is a Python implementation of the file formats and"
+                b" protocols",
             }
 
-    def createDeb(self, filename, control_format, data_format, members=None):
+    def _writeCompressedFile(self, filename, data):
+        if filename.endswith(".gz"):
+            open_func = gzip.open
+        elif filename.endswith(".xz"):
+            open_func = partial(lzma.LZMAFile, format=lzma.FORMAT_XZ)
+        else:
+            raise ValueError(
+                "Unhandled compression extension in '%s'" % filename)
+        with open_func(filename, "wb") as f:
+            f.write(data)
+
+    def createDeb(self, filename, control, control_format, data_format,
+                  members=None):
         """Return the contents of a dummy .deb file."""
         tempdir = self.makeTemporaryDirectory()
         if members is None:
@@ -374,7 +396,31 @@
                 "data.tar.%s" % data_format,
                 ]
         for member in members:
-            write_file(os.path.join(tempdir, member), b"")
+            if member == "debian-binary":
+                write_file(os.path.join(tempdir, member), b"2.0\n")
+            elif member.startswith("control.tar."):
+                with io.BytesIO() as control_tar_buf:
+                    with tarfile.open(
+                            mode="w", fileobj=control_tar_buf) as control_tar:
+                        with io.BytesIO() as control_buf:
+                            Deb822(control).dump(
+                                fd=control_buf, encoding="UTF-8")
+                            control_buf.seek(0)
+                            tarinfo = tarfile.TarInfo(name="control")
+                            tarinfo.size = len(control_buf.getvalue())
+                            control_tar.addfile(tarinfo, fileobj=control_buf)
+                    control_tar_bytes = control_tar_buf.getvalue()
+                self._writeCompressedFile(
+                    os.path.join(tempdir, member), control_tar_bytes)
+            elif member.startswith("data.tar."):
+                with io.BytesIO() as data_tar_buf:
+                    with tarfile.open(mode="w", fileobj=data_tar_buf):
+                        pass
+                    data_tar_bytes = data_tar_buf.getvalue()
+                self._writeCompressedFile(
+                    os.path.join(tempdir, member), data_tar_bytes)
+            else:
+                raise ValueError("Unhandled .deb member '%s'" % member)
         retcode = subprocess.call(
             ["ar", "rc", filename] + members, cwd=tempdir)
         self.assertEqual(0, retcode)
@@ -383,13 +429,16 @@
 
     def createDebBinaryUploadFile(self, filename, component_and_section,
                                   priority_name, package, version, changes,
-                                  control_format=None, data_format=None,
-                                  members=None):
+                                  control=None, control_format=None,
+                                  data_format=None, members=None):
         """Create a DebBinaryUploadFile."""
-        if (control_format is not None or data_format is not None or
-                members is not None):
+        if (control is not None or control_format is not None or
+                data_format is not None or members is not None):
+            if control is None:
+                control = self.getBaseControl()
             data = self.createDeb(
-                filename, control_format, data_format, members=members)
+                filename, control, control_format, data_format,
+                members=members)
         else:
             data = "DUMMY DATA"
         (path, md5, sha1, size) = self.writeUploadFile(filename, data)
@@ -415,13 +464,49 @@
         self.assertEqual("0.42", uploadfile.source_version)
         self.assertEqual("0.42", uploadfile.control_version)
 
+    def test_verifyFormat_missing_control(self):
+        # verifyFormat rejects .debs with no control member.
+        uploadfile = self.createDebBinaryUploadFile(
+            "foo_0.42_i386.deb", "main/python", "unknown", "mypkg", "0.42",
+            None, members=["debian-binary", "data.tar.gz"])
+        self.assertThat(
+            ["".join(error.args) for error in uploadfile.verifyFormat()],
+            MatchesListwise([
+                Equals(
+                    "%s: 'dpkg-deb -I' invocation failed." %
+                    uploadfile.filename),
+                MatchesRegex(
+                    r"^ \[dpkg-deb output:\] .* has premature member "
+                    r"'data\.tar\.gz'"),
+                Equals(
+                    "%s: 'dpkg-deb -c' invocation failed." %
+                    uploadfile.filename),
+                MatchesRegex(
+                    r"^ \[dpkg-deb output:\] .* has premature member "
+                    r"'data\.tar\.gz'"),
+                ]))
+
+    def test_verifyFormat_missing_data(self):
+        # verifyFormat rejects .debs with no data member.
+        uploadfile = self.createDebBinaryUploadFile(
+            "foo_0.42_i386.deb", "main/python", "unknown", "mypkg", "0.42",
+            None, members=["debian-binary", "control.tar.gz"])
+        self.assertThat(
+            ["".join(error.args) for error in uploadfile.verifyFormat()],
+            MatchesListwise([
+                Equals(
+                    "%s: 'dpkg-deb -c' invocation failed." %
+                    uploadfile.filename),
+                MatchesRegex(
+                    r"^ \[dpkg-deb output:\] .* unexpected end of file"),
+                ]))
+
     def test_verifyFormat_control_xz(self):
         # verifyFormat accepts .debs with an xz-compressed control member.
         uploadfile = self.createDebBinaryUploadFile(
             "foo_0.42_i386.deb", "main/python", "unknown", "mypkg", "0.42",
             None, control_format="xz", data_format="gz")
-        control = self.getBaseControl()
-        uploadfile.parseControl(control)
+        uploadfile.extractAndParseControl()
         self.assertEqual([], list(uploadfile.verifyFormat()))
 
     def test_verifyFormat_data_xz(self):
@@ -429,8 +514,7 @@
         uploadfile = self.createDebBinaryUploadFile(
             "foo_0.42_i386.deb", "main/python", "unknown", "mypkg", "0.42",
             None, control_format="gz", data_format="xz")
-        control = self.getBaseControl()
-        uploadfile.parseControl(control)
+        uploadfile.extractAndParseControl()
         self.assertEqual([], list(uploadfile.verifyFormat()))
 
     def test_verifyDebTimestamp_SystemError(self):
@@ -456,8 +540,8 @@
         bpr = uploadfile.storeInDatabase(build)
         self.assertEqual(u'python (<< 2.7), python (>= 2.5)', bpr.depends)
         self.assertEqual(
-            u"Dulwich is a Python implementation of the file formats "
-            u"and protocols", bpr.description)
+            u" Dulwich is a Python implementation of the file formats and"
+            u" protocols", bpr.description)
         self.assertEqual(False, bpr.essential)
         self.assertEqual(524, bpr.installedsize)
         self.assertEqual(True, bpr.architecturespecific)