Implement basic duplicate signature hash rule.

2 files changed, 63 insertions, 4 deletions

diff --git a/src/mailman_pgp/rules/signature.py b/src/mailman_pgp/rules/signature.py
index 55b9b87..395dd7d 100644
--- a/src/mailman_pgp/rules/signature.py
+++ b/src/mailman_pgp/rules/signature.py
@@ -17,6 +17,7 @@
 
 """Signature checking rule for the pgp-posting-chain."""
 from email.utils import parseaddr
+from operator import attrgetter
 
 from mailman.core.i18n import _
 from mailman.interfaces.action import Action
@@ -28,8 +29,10 @@ from zope.interface import implementer
 
 from mailman_pgp.model.address import PGPAddress
 from mailman_pgp.model.list import PGPMailingList
+from mailman_pgp.model.sighash import PGPSigHash
 from mailman_pgp.pgp.wrapper import PGPWrapper
 from mailman_pgp.utils.moderation import record_action
+from mailman_pgp.utils.pgp import hashes, verifies
 
 
 @public
@@ -89,14 +92,34 @@ class Signature:
                           'No key set for address {}.'.format(email))
             return True
 
+        if not pgp_address.key_confirmed:
+            record_action(msg, msgdata, Action.reject, email,
+                          'Key not confirmed.')
+            return True
+
+        verifications = list(wrapped.verify(key))
+
         # Take the `invalid_sig_action` if the verification failed.
-        if not wrapped.verifies(key):
+        if not verifies(verifications):
             action = pgp_list.invalid_sig_action
             if action != Action.defer:
                 record_action(msg, msgdata, action, email,
                               'Signature did not verify.')
                 return True
 
+        sig_hashes = set(hashes(verifications))
+        duplicates = set(PGPSigHash.hashes(sig_hashes))
+        if duplicates:
+            fingerprints = map(attrgetter('fingerprint'), duplicates)
+            if key.fingerprint in fingerprints:
+                action = pgp_list.duplicate_sig_action
+                if action != Action.defer:
+                    record_action(msg, msgdata, action, email,
+                                  'Signature duplicate.')
+                    return True
+
+        # TODO: add the sig hashes to the db.
+
         # XXX: we need to track key revocation separately to use it here
         # TODO: check key revocation here
 
diff --git a/src/mailman_pgp/rules/tests/test_signature.py b/src/mailman_pgp/rules/tests/test_signature.py
index f5c5dc3..d8d0537 100644
--- a/src/mailman_pgp/rules/tests/test_signature.py
+++ b/src/mailman_pgp/rules/tests/test_signature.py
@@ -28,9 +28,12 @@ from mailman_pgp.config import mm_config
 from mailman_pgp.database import mm_transaction, transaction
 from mailman_pgp.model.address import PGPAddress
 from mailman_pgp.model.list import PGPMailingList
+from mailman_pgp.model.sighash import PGPSigHash
 from mailman_pgp.pgp.tests.base import load_key, load_message
+from mailman_pgp.pgp.wrapper import PGPWrapper
 from mailman_pgp.rules.signature import Signature
 from mailman_pgp.testing.layers import PGPConfigLayer
+from mailman_pgp.utils.pgp import hashes
 
 
 class TestSignatureRule(TestCase):
@@ -49,10 +52,11 @@ class TestSignatureRule(TestCase):
 
         self.pgp_list = PGPMailingList.for_list(self.mlist)
 
-        sender_key = load_key('rsa_1024.pub.asc')
+        self.sender_key = load_key('rsa_1024.priv.asc')
         with transaction() as t:
             self.pgp_sender = PGPAddress(self.sender.preferred_address)
-            self.pgp_sender.key = sender_key
+            self.pgp_sender.key = self.sender_key.pubkey
+            self.pgp_sender.key_confirmed = True
             t.add(self.pgp_sender)
 
         self.msg_clear = load_message('clear.eml')
@@ -98,7 +102,7 @@ To: test@example.com
         self.assertTrue(matches)
         self.assertAction(msgdata, Action.reject, [
             'No key set for address {}.'.format(
-                self.pgp_sender.address.original_email)])
+                    self.pgp_sender.address.original_email)])
 
     def assertAction(self, msgdata, action, reasons):
         self.assertEqual(msgdata['moderation_action'], action.name)
@@ -111,6 +115,7 @@ To: test@example.com
             self.pgp_list.expired_sig_action = Action.defer
             self.pgp_list.invalid_sig_action = Action.defer
             self.pgp_list.revoked_sig_action = Action.defer
+            self.pgp_list.duplicate_sig_action = Action.defer
 
         msgdata = {}
         matches = self.rule.check(self.mlist, self.msg_clear, msgdata)
@@ -130,6 +135,7 @@ To: test@example.com
             self.pgp_list.expired_sig_action = Action.defer
             self.pgp_list.invalid_sig_action = Action.defer
             self.pgp_list.revoked_sig_action = Action.defer
+            self.pgp_list.duplicate_sig_action = Action.defer
 
         msgdata = {}
         matches = self.rule.check(self.mlist, self.msg_inline_signed, msgdata)
@@ -146,6 +152,7 @@ To: test@example.com
             self.pgp_list.expired_sig_action = Action.defer
             self.pgp_list.invalid_sig_action = Action.hold
             self.pgp_list.revoked_sig_action = Action.defer
+            self.pgp_list.duplicate_sig_action = Action.defer
 
         msgdata = {}
         matches = self.rule.check(self.mlist, self.msg_inline_signed_invalid,
@@ -158,3 +165,32 @@ To: test@example.com
                                   msgdata)
         self.assertTrue(matches)
         self.assertAction(msgdata, Action.hold, ['Signature did not verify.'])
+
+    def test_duplicate_sig_action(self):
+        with transaction() as t:
+            self.pgp_list.unsigned_msg_action = Action.defer
+            self.pgp_list.inline_pgp_action = Action.defer
+            self.pgp_list.expired_sig_action = Action.defer
+            self.pgp_list.invalid_sig_action = Action.defer
+            self.pgp_list.revoked_sig_action = Action.defer
+            self.pgp_list.duplicate_sig_action = Action.hold
+
+            wrapped = PGPWrapper(self.msg_mime_signed)
+            sig_hashes = set(hashes(wrapped.verify(self.sender_key.pubkey)))
+            wrapped = PGPWrapper(self.msg_inline_signed)
+            sig_hashes |= set(hashes(wrapped.verify(self.sender_key.pubkey)))
+            for hash in sig_hashes:
+                sig_hash = PGPSigHash()
+                sig_hash.hash = hash
+                sig_hash.fingerprint = self.sender_key.pubkey.fingerprint
+                t.add(sig_hash)
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_mime_signed, msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['Signature duplicate.'])
+
+        msgdata = {}
+        matches = self.rule.check(self.mlist, self.msg_inline_signed, msgdata)
+        self.assertTrue(matches)
+        self.assertAction(msgdata, Action.hold, ['Signature duplicate.'])