piknik

commit 6cdb76451876b1b46fe59aa8e422c2e9ab449d22
parent 6c27c2206c000db151acde595657ecde809cb652
Author: lash <dev@holbrook.no>
Date:   Wed,  9 Nov 2022 15:11:34 +0000

Add verify signature test

Diffstat:
M
piknik/msg.py
 | 
36
+++++++++++++++++++-----------------
M
tests/test_crypto.py
 | 
24
+++++++++++++++++++-----
M
tests/test_store.py
 | 
3
+--

3 files changed, 39 insertions(+), 24 deletions(-)
diff --git a/piknik/msg.py b/piknik/msg.py
@@ -35,34 +35,34 @@ class IssueMessage:
         return o
 
 
-    def add_text(self, m, v):
-        p = Message()
-        p.add_header('Content-Transfer-Encoding', 'QUOTED-PRINTABLE')
-        p.set_charset('UTF-8')
-        p.set_payload(str(v))
-        m.attach(p)
+    def from_text(self, v):
+        m = Message()
+        m.add_header('Content-Transfer-Encoding', 'QUOTED-PRINTABLE')
+        m.set_charset('UTF-8')
+        m.set_payload(str(v))
+        return m
 
 
     def detect_file(self, v):
         return mimetypes.guess_type(v)
 
 
-    def add_file(self, m, v):
+    def from_file(self, v):
         mime_type = self.detect_file(v)
 
-        p = Message()
-        p.set_type(mime_type[0])
+        m = Message()
+        m.set_type(mime_type[0])
         if mime_type[1] != None:
-            p.set_charset(mime-type[1])
-        p.add_header('Content-Transfer-Encoding', 'BASE64')
+            m.set_charset(mime-type[1])
+        m.add_header('Content-Transfer-Encoding', 'BASE64')
 
         f = open(v, 'rb')
         r = f.read()
         f.close()
         r = b64encode(r)
+        m.set_payload(str(r))
 
-        p.set_payload(str(r))
-        m.attach(p)
+        return m
 
 
     def add(self, *args, related_id=None, wrapper=None):
@@ -78,12 +78,14 @@ class IssueMessage:
         for a in args:
             p = a[:2]
             v = a[2:]
+            r = None
             if p == 'f:':
-                self.add_file(m, v)
+                r = self.from_file(v)
             elif p == 's:':
-                self.add_text(m, v)
-        if wrapper:
-            m = wrapper(m)
+                r = self.from_text(v)
+            if wrapper:
+                r = wrapper(r)
+            m.attach(r)
         self.__m.attach(m)
 
 
diff --git a/tests/test_crypto.py b/tests/test_crypto.py
@@ -30,9 +30,9 @@ class TestMsg(unittest.TestCase):
     def setUp(self):
         self.store = TestStates()
         self.gpg_dir = tempfile.mkdtemp()
-        gpg = gnupg.GPG(gnupghome=self.gpg_dir)
-        gpg_input = gpg.gen_key_input(key_type='RSA', key_length=1024, passphrase='foo')
-        gpg_key = gpg.gen_key(gpg_input)
+        self.gpg = gnupg.GPG(gnupghome=self.gpg_dir)
+        gpg_input = self.gpg.gen_key_input(key_type='RSA', key_length=1024, passphrase='foo')
+        gpg_key = self.gpg.gen_key(gpg_input)
         self.crypto = PGPSigner(self.gpg_dir, default_key=gpg_key.fingerprint, passphrase='foo')
         self.b = Basket(self.store, message_wrapper=self.crypto.sign)
 
@@ -44,8 +44,22 @@ class TestMsg(unittest.TestCase):
         m = Message()
         m.set_charset('utf-8')
         m.set_payload('foo')
-        r = self.crypto.sign(m, passphrase='foo')
-        print(str(r))
+        v = m.as_string()
+        m = self.crypto.sign(m, passphrase='foo')
+
+        print('msg {}'.format(v))
+        for p in m.walk():
+            if p.get_content_type() == 'application/pgp-signature':
+                sig = p.get_payload()
+                (fd, fp) = tempfile.mkstemp()
+                f = os.fdopen(fd, 'w')
+                f.write(sig)
+                f.close()
+                r = self.gpg.verify_data(fp, v.encode('utf-8'))
+                os.unlink(fp)
+                self.assertIsNone(r.key_status)
+                self.assertEqual(r.status, 'signature valid')
+                break
 
 
     def test_wrap_basket_sig(self):
diff --git a/tests/test_store.py b/tests/test_store.py
@@ -81,8 +81,7 @@ class TestStore(unittest.TestCase):
     def test_msg_putget(self):
         o = Issue('foo')
         issue_id = self.b.add(o)
-        m = self.b.msg(issue_id, 'bar')
-
+        m = self.b.msg(issue_id, 's:bar')
 
 
     def test_msg_resume(self):