[PATCH] Refactor GPG code; add signature verification and relevant test cases.

Serafeim Zanikolas serzan at hellug.gr
Sun Feb 8 00:22:44 CET 2009


---
 debexpo/controllers/my.py                   |    4 +-
 debexpo/lib/utils.py                        |  110 ++++++++++++++++++++++-----
 debexpo/lib/validators.py                   |    4 +-
 debexpo/tests/gpg/pubring_with_355304E4.gpg |  Bin 0 -> 1448 bytes
 debexpo/tests/gpg/signed_by_355304E4.gpg    |  Bin 0 -> 445 bytes
 debexpo/tests/test_utils.py                 |   93 ++++++++++++++++++++++
 6 files changed, 188 insertions(+), 23 deletions(-)
 create mode 100644 debexpo/tests/gpg/pubring_with_355304E4.gpg
 create mode 100644 debexpo/tests/gpg/signed_by_355304E4.gpg

diff --git a/debexpo/controllers/my.py b/debexpo/controllers/my.py
index eea715f..dae9ba0 100644
--- a/debexpo/controllers/my.py
+++ b/debexpo/controllers/my.py
@@ -41,7 +41,7 @@ import md5
 from debexpo.lib.base import *
 from debexpo.lib import constants, form
 from debexpo.lib.schemas import DetailsForm, GpgForm, PasswordForm, OtherDetailsForm
-from debexpo.lib.utils import parse_key_id
+from debexpo.lib.utils import gnupg
 
 from debexpo.model import meta
 from debexpo.model.users import User
@@ -99,7 +99,7 @@ class MyController(BaseController):
         if 'gpg' in self.form_result and self.form_result['gpg'] is not None:
             log.debug('Setting a new GPG key')
             self.user.gpg = self.form_result['gpg'].value
-            self.user.gpg_id = parse_key_id(self.user.gpg)
+            self.user.gpg_id = gnupg.parse_key_id(self.user.gpg)
 
         meta.session.commit()
 
diff --git a/debexpo/lib/utils.py b/debexpo/lib/utils.py
index 6bec930..31c58b1 100644
--- a/debexpo/lib/utils.py
+++ b/debexpo/lib/utils.py
@@ -112,22 +112,94 @@ def md5sum(filename):
 
     return sum.hexdigest()
 
-def parse_key_id(key):
-    """
-    Returns the key id of the given GPG public key.
-
-    ``key``
-        ASCII armored GPG public key.
-    """
-    gpg_path = config['debexpo.gpg_path']
-    if not gpg_path:
-        log.error('debexpo.gpg_path is not set in configuration file.')
-        return None
-    cmd = 'echo "%s" | %s' % (key, gpg_path)
-    status, output = commands.getstatusoutput(cmd)
-    if status != 0:
-        return None
-    try:
-        return output.split()[1]
-    except KeyError:
-        return None
+class GnuPG:
+
+    GPG_PATH_NOT_INITIALISED = -1
+    INVALID_GNUPG_RUN_INVOCATION = -2
+
+    def __init__(self):
+        """
+        Wrapper for certain GPG operations.
+
+        Meant to be instantiated only once.
+        """
+        try:
+            self.gpg_path = config['debexpo.gpg_path']
+        except KeyError:
+            self.gpg_path = None
+        if self.gpg_path is None:
+            log.error('debexpo.gpg_path is not set in configuration file' +
+                      ' (or is set to a blank value)')
+        elif not os.path.isfile(self.gpg_path):
+            log.error('debexpo.gpg_path refers to a non-existent file')
+            self.gpg_path = None
+        elif not os.access(self.gpg_path, os.X_OK):
+            log.error('debexpo.gpg_path refers to a non-executable file')
+            self.gpg_path = None
+
+    def parse_key_id(self, key):
+        """
+        Returns the key id of the given GPG public key.
+
+        ``key``
+            ASCII armored GPG public key.
+
+        Sample output to be parsed:
+
+        pub  1024D/355304E4 2005-09-13 Serafeim Zanikolas <serzan at hellug.gr>
+        sub  1024g/C082E9B7 2005-09-13 [expires: 2008-09-12]
+
+        """
+        try:
+            (output, _) = self._run(stdin_cmd="echo '%s'" % key)
+            return output.split()[1] # get only the 2nd column of the 1st line
+        except (AttributeError, IndexError):
+            log.error('Failed to extract key id from gpg output: "%s"'
+                       % output)
+
+    def verify_sig(self, signed_file, pubring=None):
+        """
+        Returns true if the given GPG-signed file can be verified.
+
+        ```signed_file```
+             path to signed file
+        ```pubring```
+             path to public key ring (when not specified, the default GPG
+             setting will be used (~/.gnupg/pubring.gpg))
+        """
+        if pubring is None:
+            args = "--verify"
+        else:
+            args = "--verify --keyring %s --no-default-keyring" % pubring
+        (_, status) = self._run(stdin_cmd="cat %s" % signed_file, args=args)
+        return status == 0
+
+    def _run(self, stdin_cmd=None, args=None):
+        """
+        Run gpg with the given stdin and arguments and return the output and
+        exit status.
+
+        ``stdin_cmd``
+            run this command and pipe its output to gpg
+        ``args``
+            a string to be passed as argument(s) to gpg
+
+        """
+        if self.gpg_path is None:
+            return (None, GnuPG.GPG_PATH_NOT_INITIALISED)
+        if stdin_cmd and args:
+            cmd = '%s | %s %s' % (stdin_cmd, self.gpg_path, args)
+        elif stdin_cmd:
+            cmd = '%s | %s' % (stdin_cmd, self.gpg_path)
+        elif args:
+            cmd = '%s %s' % (self.gpg_path, args)
+        else:
+            log.error('GnuPG._run() invoked without any arguments.')
+            return (None, GnuPG.INVALID_GNUPG_RUN_INVOCATION)
+        status, output = commands.getstatusoutput(cmd)
+        open("/tmp/blah", "w").write(cmd + "\n\n" + str(status))
+        if status != 0:
+            return (None, status)
+        return (output, 0)
+
+gnupg = GnuPG()
diff --git a/debexpo/lib/validators.py b/debexpo/lib/validators.py
index 7c191ad..9c50c45 100644
--- a/debexpo/lib/validators.py
+++ b/debexpo/lib/validators.py
@@ -40,7 +40,7 @@ import logging
 import md5
 
 from debexpo.lib.base import *
-from debexpo.lib.utils import parse_key_id
+from debexpo.lib.utils import gnupg
 
 from debexpo.model import meta
 from debexpo.model.users import User
@@ -69,7 +69,7 @@ class GpgKey(formencode.validators.FieldStorageUploadConverter):
             log.error('GPG key does not start with BEGIN PGP PUBLIC KEY BLOCK')
             raise formencode.Invalid(_('Invalid GPG key'), value, c)
 
-        self.gpg_id = parse_key_id(value.value)
+        self.gpg_id = gnupg.parse_key_id(value.value)
         if self.gpg_id is None:
             log.error("Failed to parse GPG key")
             raise formencode.Invalid(_('Invalid GPG key'), value, c)
diff --git a/debexpo/tests/gpg/pubring_with_355304E4.gpg b/debexpo/tests/gpg/pubring_with_355304E4.gpg
new file mode 100644
index 0000000000000000000000000000000000000000..576d49973f0083fdfa8427015a5d41ae31232eab
GIT binary patch
literal 1448
zcmZ{ki96JJ9LK+NnHl#rDN~L`Wi!e|TWQA?Lg`^9WTd9TBu6HxNvtDgvRL;uHS=>l
z*fFh2DY=G}iA}T;I+3Gj!fKw3V~ySCS<gQ0Kk#}z-_QH?eDh&NNV`q=1T`cWYj~LO
zjMXJCv}nmdJ364tLi5ji7 at SWo?}dKa)YmsmpL4+!5=sjhW-ebJf3!M%@`F3;h;9Xq
zJ022jko4LNcBZLM6WLHoFW~I1&$hN5aiBy;iZS+GMtsXz(M0;(=RV7&wr!m^)hVLD
z-oUf%LmD`%h+h79p!!ahvsetE at f2r!4@%j9I*xN=)AhxqImIYD#G)Z$dLi}+W4tF!
zr at J=oWeq1SkXbwIfth80jN%HXNk_Mb`OfienR9gRs6$)r=t!sf)umsFc{h#K6=2Ew
z{<R{aQtfl){zyyH-dpV{N5zp3!uuLVT$KY7gi>?y!}wf5px;Dm?#oZ!wQhX0bTn(Y
zrKm(K)mF+_U|Y$nAQpc#t?%Se&I$UFa(Mi;m at 1ULfUKjZCTtqwN%4d&1*IYKx)T-e
za|W49yaz`W%n%{yDNf71l!5s4ca;}IV+?s0)$p<2o3jJT<|-rHgi*aC_o?d&ZMx}s
zL(RQtM5rddHn`n_uz9G(ZJw1+J(KP2B#Ov$lKhDpxyo?PtTrP1F=+mNr-JpoX(6Xh
zhX&Ek>g_q}$GAufv8MY41)U2p4PcPiG^B<a9O%m_U@>R}9D#!)5NqHtxDE`i6@!BB
z1DQ4`tfH at d9gxSWPBz;gw;&^jCz$}MN{?O|rjP at _Mx{+0=$H~++6ngoP!_71FHV~@
z-byVlF)tjv>+WFsstOJeZ0d>z|34bXW}INPfK?2*8z=FCAJ(^fF3k5uw^k=oal)w^
z0D9DRr^G08<A>;M`u)9<l36klfmsp!^9~eTRwfFIDlUlmrfjt)wGQwOrW|U^qIK(g
z9UgiJSmf3w!>}p<^}n4ymaL)aZ{icxRstF14tnnLUak2iT6V5q4#>`B)3#b;XaS%e
z36r4W;Eq3;(O$p%z1^C)IGzMR^<>f*>=P_k89rJdjC~#?S#pS9(KsMgmsgQnCThen
zWh|UmmVN%%2I!3efbNbm%^#(2la96w-GUWiYnq&x*?`BqlLtOr_iTRg^=7nyv1tA}
z({<l63$XuXDRgd^v1;5jvGcfriH7R{;`puINJ~&9c<$db?@zkzvHiql7J$06T`za!
zT)5{pUE1?`q0~#|NVwyFSe9Q_7Rxc|msmc)=u(rwwe at +q?df-{NnCSLsI@)U2IrcH
z1yJG}1>g6nb&1Z})Z}VSM at F*-M<iq2gDJ^=5tb?vq+XdJ&2rD+>z+z|XT(mQ-RKri
z$Q!~OFP|e76s2gp4v??dGAa8<20HfeUU at e1^U9U9;;|DQI(Eo4r2 at I{YhtDE^pc2K
z9s_EUf}l(yRliOVh=%=cVSIF at x~;Z}+a8Tq2M1`nmZ_NE9+rP?W{y`uzm#jtcvi~8
z=$6j_1PU>)Ozis99DB;P(snR>%igjtOhpmJaBCIQF8{HUh)?)Ez8l&{GW_F`zgFsR
zQ$JJYwcs-siu13Eisc%u7Qc8A2 at 2^((A2vfk&U~Qq70|D>g6BIu~s}0v13eO^9B%l
zGPhFF^LCw_F{e0^dpznf+ZVWF3|<{O^rFH9)41SL#@>%yyS%yLO2MqZ3tnAOjpgz#
v;T|4#6;BV(CiZ!X5Vy^qn%p1&o)6!+9Q~r;v`Y{R`CV_gga}Jz%PRT{o6B6z

literal 0
HcmV?d00001

diff --git a/debexpo/tests/gpg/signed_by_355304E4.gpg b/debexpo/tests/gpg/signed_by_355304E4.gpg
new file mode 100644
index 0000000000000000000000000000000000000000..73293aae937d0f9c46d167898d5493a4954713c7
GIT binary patch
literal 445
zcmV;u0Yd(x0nL;@OT$1I#$&Dj0)lRW%TbpUYN1sTOT|SC1#wYaN-1~ETXRq^;qHp<
z>L5<y<f^VpzkxV(aB}byln$;AegMHslg9Lqpo{oU;mPyle)rzJxttx>BJ*#p9kR9`
zeI08?Q)HS*5HjgB@)wJfx(O9jwvP#AP!hl?2t%Y2NTlN#9jU at -6NyxkL0ua0u0mWT
z9aeD3v>7NlhpPfS)39`6(JCNjOJw)~%uL_tFXk4lbUh)9;eX8LiYALl$m|veq1M5L
zOtDXvywUF)y_;S^y$p^-`_gvu5;&IBW+U>wZ5{9$1aG10Lz56R4xovG#v`H0kD!Me
z{yS*$4YYuM?-Fu=Ti6l)QK at Dr#I1kU>_2W(gF at AGx2<<a<p@|wO$aah=UGwbOA%#A
zNtuBV^&-cZG9xrig&5EyOU4q&djMduVlY{SFI8H0O{lvRIIS8pIhZlz+0~LMmGF>o
zyr-oDjl}yI2j^Vn6Wv7Uaypzc>7sksx~`w+qS|~!o1BQaKMYNund|*dI*;1Ta~rK+
n-rC!HhYy{*x$Jvu%3RQHu2vII?Typ><!jhoF+L9J&9j$Jxgy!J

literal 0
HcmV?d00001

diff --git a/debexpo/tests/test_utils.py b/debexpo/tests/test_utils.py
index eb04fd0..1a021b0 100644
--- a/debexpo/tests/test_utils.py
+++ b/debexpo/tests/test_utils.py
@@ -36,10 +36,51 @@ __copyright__ = 'Copyright © 2008 Jonny Lamb'
 __license__ = 'MIT'
 
 from unittest import TestCase
+import os
 
 from debexpo.lib.utils import *
 from debexpo.lib.changes import Changes
 
+test_gpg_key = \
+"""-----BEGIN PGP PUBLIC KEY BLOCK-----
+Version: GnuPG v2.0.9 (GNU/Linux)
+
+mQGiBEMnG4URBADovsaF04fRCsE1w5IHR0YHp2+Zd1Yjd4yo16B/J9nZ5Gj6Swih
+LaWhcjFL+crrP2tk6lKHXR+pYZ7pbm0jit4xAXDA2RQEvqVomps6vZVAQuZGVH57
+7whF0SWrO/XJ7JH68Nk7/8gwz7ISVMlq12pzy+MTFT9216vpahI4h0rv7wCg6Y1K
+RVZUp9sSFZuxJ67+ivoMfUMD/iQD8v2BznLp1XEe0rqQ3LebkGp5uuRWCPWI632e
+wfI+XzNxXvqrQnn6aJ7nRsi65+bPEpz/qjKYsikSCwGMIWa6yTINutYO2ns7Ltez
+y41f73vEdNm+5k4OZ4XB+zTvxoOXrWpl7MWX3O5aulGrB/vnlOTDoqTv+xoNkv2I
+PwoPA/49Lr3Pm1R1rdoEBhsbnYCwBUWtUx4gEcHA45/7Gy0rmqVuCh/sqeWW4nn/
+n8RfCzEDbgfxm8O5jduDkeGsem+AJJ96ERuBWiiVZ6f6rHZRwX3X5rtbGaFB0miY
+48LXBwNvFBu6bcs1LSjjw1H8h5lbcJVaScl2mEn39AXmnHJKk7QlU2VyYWZlaW0g
+WmFuaWtvbGFzIDxzZXJ6YW5AaGVsbHVnLmdyPohhBBMRAgAhBgsJCAcDAgMVAgMD
+FgIBAh4BAheABQJIAJNBBQkMXkW8AAoJEO3CRWI1UwTk7XkAoLCRfv/kVFNq+X2Q
+7E3M8cl8OAJcAKCUBbSr75DtKS9bo6Q0oeK4UkYu3YhbBBMRAgAbBgsJCAcDAgMV
+AgMDFgIBAh4BAheABQJHwhWtAAoJEO3CRWI1UwTkh6MAnibtG603HMtX/fzZfsW0
+hlsVwfGxAKDHyLakJZMm6n6VaLtE96T1UzIDCIhhBBMRAgAhBQJDJxuFBQkFo5qA
+BgsJCAcDAgMVAgMDFgIBAh4BAheAAAoJEO3CRWI1UwTkjrwAn0+NVciUYdIhWFnj
+xgCHU8XAJHGwAKDa4PJgjBMUZixcfcikoCOX4lc5WohhBBMRAgAhBgsJCAcDAgMV
+AgMDFgIBAh4BAheABQJIAI9NBQkMXkHIAAoJEO3CRWI1UwTkJDUAoNX3eS0PRlIb
+ZJLLvTrlQxaCgp/3AKC9Uz7oAe4Blw4C55rBgdZs9/9Gg4hGBBARAgAGBQJIE4J5
+AAoJEBVYlEWZ6B2g3yMAoLjneTTHkTD758PjswGiCbfASXmVAJ95tpgA6q5Xwtj5
+sn6tcv403pNOSIhGBBARAgAGBQJIC88zAAoJELdRFAn8Fdvsvq4AoJtlGCZhhRAt
+V0w8/GY+tVYzY4SHAKDRGk6EzJZ4uVHypdXw/aVYD110R4hGBBARAgAGBQJIO/at
+AAoJEJYs2vc7xAgfW4oAnRyYl8uRtkA+njTJb0BFnkEVToYJAKCG3wte5Y68hkoa
+W4y0FEdywhObybkBDQRDJxuHEAQAjonzPvWecBu80Pte8+9J8FFoNc5THXFHhHU+
+mqKNGk7bU4lCeVRM5tvMPJ/dV7+rmKgNF4MJ7MweQwQWpa0GKreB++EgijKUVtsR
+95pskzJbIbwMAMnkZbMIXB/7H8VChjDH6bRtZxROpw80teQK3jE0Gw8H3Aa/ktOl
+nwgfqPMAAwUD/A4y0e7CgWlCrELidCtEp/Z5DMlUJC+weUOZyknJqy3Ng9KgSD4k
+1HxmF46v8YtU/BcC83ijmZzJowa/P/72WDItC5EloPHhNnu/OQ19JPEvIJlPlkAM
+Y3Y26AsoHQBvZJes99XgGQYpm6N7nmJ9yoheAFIII91gVdipLAi//UuniEwEGBEC
+AAwFAkMnG4cFCQWjmoAACgkQ7cJFYjVTBOS3OwCg0XRWVkOp0Fn1htlXyQO1MdAs
+sS0An1yrKagH2JprS2yHBCLXdPcyAY6I
+=VNMB
+-----END PGP PUBLIC KEY BLOCK-----
+"""
+
+test_gpg_key_id = "1024D/355304E4"
+
 class TestUtilsController(TestCase):
 
     def testAllowedUpload(self):
@@ -80,3 +121,55 @@ class TestUtilsController(TestCase):
         Tests debexpo.lib.utils.md5sum.
         """
         self.assertEqual(md5sum('debexpo/tests/changes/synce-hal_0.1-1_source.changes'), 'fbb0b9c81f8a4fa9b8e3b789cf3b5220')
+
+    def _get_gnupg(self, gpg_path='/usr/bin/gpg'):
+        config['debexpo.gpg_path'] = gpg_path
+        gnupg = GnuPG() # instantiate with new debexpo.gpg_path setting
+        return gnupg
+
+    def testGnuPGfailure1(self):
+        """
+        Test for debexpo.gpg_path being uninitialised.
+        """
+        gnupg = self._get_gnupg(None)
+        self.assertEqual(gnupg.gpg_path, None)
+        
+    def testGnuPGfailure2(self):
+        """
+        Test for debexpo.gpg_path pointing to a non-existent file.
+        """
+        gnupg = self._get_gnupg('/non/existent')
+        self.assertEqual(gnupg.gpg_path, None)
+        
+    def testGnuPGfailure3(self):
+        """
+        Test for debexpo.gpg_path pointing to a non-executable file.
+        """
+        gnupg = self._get_gnupg('/etc/passwd')
+        self.assertEqual(gnupg.gpg_path, None)
+        
+    def testParseKeyID(self):
+        """
+        Test the extraction of key id from a given GPG key.
+        """
+        gnupg = self._get_gnupg()
+        self.assertEqual(gnupg.parse_key_id(test_gpg_key), test_gpg_key_id)
+
+    def testSignatureVerification(self):
+        """
+        Verify the signature in the file debexpo/tests/gpg/signed_by_355304E4.
+        """
+        gnupg = self._get_gnupg()
+        gpg_data_dir = os.path.join(os.path.dirname(__file__), "gpg")
+        signed_file = os.path.join(gpg_data_dir, "signed_by_355304E4.gpg")
+        pubring = os.path.join(gpg_data_dir, "pubring_with_355304E4.gpg")
+        assert os.path.exists(signed_file)
+        assert os.path.exists(pubring)
+        self.assertTrue(gnupg.verify_sig(signed_file, pubring))
+
+    def testInvalidSignature(self):
+        """
+        Test that verify_sig() fails for an unsigned file.
+        """
+        gnupg = self._get_gnupg()
+        self.assertFalse(gnupg.verify_sig("/etc/passwd"))
-- 
1.5.6.5


--XOIedfhf+7KOe/yw--


More information about the debexpo-devel mailing list