Bug#485829: python-debian: pgp signature checking for deb822

[Filippo Giunchedi]

Package: python-debian
Version: 0.1.10
Severity: wishlist
Tags: patch

Hi,
as per request on pkg-python-debian-discuss at lists.alioth.debian.org I'm
issuing a bug report for initial support of pgp signature checking for
deb822.

thanks,
filippo

-- System Information:
Debian Release: lenny/sid
  APT prefers unstable
  APT policy: (990, 'unstable')
Architecture: amd64 (x86_64)

Kernel: Linux 2.6.25.1-mactel (SMP w/2 CPU cores)
Locale: LANG=en_US, LC_CTYPE=en_US (charmap=ISO-8859-1)
Shell: /bin/sh linked to /bin/bash

Versions of packages python-debian depends on:
ii  python                        2.5.2-1    An interactive high-level object-o
ii  python-support                0.8.1      automated rebuilding support for P

Versions of packages python-debian recommends:
ii  python-apt                    0.7.5      Python interface to libapt-pkg

-- no debconf information
-------------- next part --------------
=== modified file 'debian_bundle/deb822.py'
--- debian_bundle/deb822.py	2008-06-07 09:00:50 +0000
+++ debian_bundle/deb822.py	2008-06-11 17:52:35 +0000
@@ -161,6 +161,8 @@
             except EOFError:
                 pass
 
+        self.pgp_info = None
+
     def iter_paragraphs(cls, sequence, fields=None, shared_storage=True):
         """Generator that yields a Deb822 object for each paragraph in sequence.
 
@@ -388,8 +390,121 @@
 
     gpg_stripped_paragraph = staticmethod(gpg_stripped_paragraph)
 
+    def get_pgp_info(self):
+        """Return a GpgInfo object with GPG signature information
+
+        This method will raise ValueError if the signature is not available
+        (e.g. the original text cannot be found)"""
+
+        # raw_text is saved (as a string) only for Changes and Dsc (see
+        # __init__ for these) which are small compared to Packages or Sources
+        # and contain no signature
+        if not hasattr(self, 'raw_text'):
+            # XXX better wording
+            raise ValueError, "raw text cannot be found"
+
+        if self.pgp_info is None:
+            self.pgp_info = GpgInfo.from_sequence(self.raw_text)
+
+        return self.pgp_info
+
 ###
 
+# XXX check what happens if input contains more that one signature
+class GpgInfo(dict):
+    """A wrapper around gnupg parsable output obtained via --status-fd
+
+    This class is really a dictionary containing parsed output from gnupg plus
+    some methods to make sense of the data.
+    Keys are keywords and values are arguments suitably splitted.
+    See /usr/share/doc/gnupg/DETAILS.gz"""
+
+    # keys with format "key keyid uid"
+    uidkeys = ('GOODSIG', 'EXPSIG', 'EXPKEYSIG', 'REVKEYSIG', 'BADSIG')
+
+    def valid(self):
+        """Is the signature valid?"""
+        return self.has_key('GOODSIG') or self.has_key('VALIDSIG')
+    
+# XXX implement as a property?
+# XXX handle utf-8 %-encoding
+    def uid(self):
+        """Return the primary ID of the signee key, None is not available"""
+        pass
+
+
+    def from_output(out, err=None):
+        """Create a new GpgInfo object from gpg(v) --status-fd output (out) and
+        optionally collect stderr as well (err).
+        
+        Both out and err can be lines in newline-terminated sequence or regular strings."""
+
+        n = GpgInfo()
+
+        if isinstance(out, basestring):
+            out = out.split('\n')
+        if isinstance(err, basestring):
+            err = err.split('\n')
+
+        n.out = out
+        n.err = err
+        
+        header = '[GNUPG:] '
+        for l in out:
+            if not l.startswith(header):
+                continue
+
+            l = l[len(header):]
+            l = l.strip('\n')
+
+            # str.partition() would be better, 2.5 only though
+            s = l.find(' ')
+            key = l[:s]
+            if key in GpgInfo.uidkeys:
+                # value is "keyid UID", don't split UID
+                value = l[s+1:].split(' ', 1)
+            else:
+                value = l[s+1:].split(' ')
+
+            n[key] = value
+        return n 
+
+    from_output = staticmethod(from_output)
+
+# XXX note that to pass additional arguments executable can be a list
+    def from_sequence(sequence, keyrings=['/usr/share/keyrings/debian-keyring.gpg'], executable=["/usr/bin/gpgv"]):
+        """Create a new GpgInfo object from the given sequence.
+
+        Sequence is a sequence of lines newline-terminated (as those returned
+        by an open file) or a regular string"""
+
+        # XXX check for gpg as well and use --verify accordingly?
+        # XXX what about --no-default-keyring?
+        args = executable
+        args.extend(["--status-fd", "1"])
+        import os
+        [args.extend(["--keyring", k]) for k in keyrings if os.path.isfile(k) and os.access(k, os.R_OK)]
+
+        import subprocess
+        p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        # XXX what to do with exit code?
+
+        if isinstance(sequence, basestring):
+            (out, err) = p.communicate(sequence)
+        else:
+            (out, err) = p.communicate("".join(sequence))
+       
+        return GpgInfo.from_output(out, err)
+
+    from_sequence = staticmethod(from_sequence)
+
+    def from_file(target, *args):
+        """Create a new GpgInfo object from the given file, calls from_string(file(target), *args)"""
+        return from_string(file(target), *args)
+    
+    from_file = staticmethod(from_file)
+
+
 class _multivalued(Deb822):
     """A class with (R/W) support for multivalued fields.
     
@@ -476,6 +591,14 @@
         "checksums-sha256": ["sha256", "size", "name"],
     }
 
+    def __init__(self, *args, **kwargs):
+        if isinstance(args[0], basestring):
+            self.raw_text = args[0]
+        else:
+            self.raw_text = "".join(args[0])
+        args = (self.raw_text,) + args[1:]
+
+        _multivalued.__init__(self, *args, **kwargs)
 
 class Changes(_multivalued):
     _multivalued_fields = {
@@ -484,6 +607,15 @@
         "checksums-sha256": ["sha256", "size", "section", "priority", "name"],
     }
 
+    def __init__(self, *args, **kwargs):
+        if isinstance(args[0], basestring):
+            self.raw_text = args[0]
+        else:
+            self.raw_text = "\n".join(args[0])
+        args = (self.raw_text,) + args[1:]
+
+        _multivalued.__init__(self, *args, **kwargs)
+
     def get_pool_path(self):
         """Return the path in the pool where the files would be installed"""