From a0ae8c24c58a61919798b45635eb9877784c20b2 Mon Sep 17 00:00:00 2001
From: Scott Kitterman <scott@kitterman.com>
Date: Fri, 18 May 2018 19:42:46 -0400
Subject: [PATCH] Detect unknown algorithm in k= tag and raise an appropriate
 error, vice       failing with a traceback

---
 ChangeLog                |  2 ++
 dkim/__init__.py         |  2 ++
 dkim/tests/data/badk.txt |  1 +
 dkim/tests/test_dkim.py  | 34 ++++++++++++++++++++++++++++++++++
 4 files changed, 39 insertions(+)
 create mode 100644 dkim/tests/data/badk.txt

diff --git a/ChangeLog b/ChangeLog
index 84223bc..09aedd2 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -6,6 +6,8 @@ Unreleased Version 0.8.0
     - Fix typo in dknewky(1) for k= tag (Thanks to Andreas Schulze for
       reporting)
     - Detect incorrect version in DKIM public key record (LP: #1763815)
+    - Detect unknown algorithm in k= tag and raise an appropriate error, vice
+      failing with a traceback
     - Indicate that ed25519-sha256 is no longer experimental
 
 2018-02-17 Version 0.7.1
diff --git a/dkim/__init__.py b/dkim/__init__.py
index c0d394a..a526a6c 100644
--- a/dkim/__init__.py
+++ b/dkim/__init__.py
@@ -406,6 +406,8 @@ def load_pk_from_dns(name, dnsfunc=get_txt):
       except (TypeError,UnparsableKeyError) as e:
           raise KeyFormatError("could not parse public key (%s): %s" % (pub[b'p'],e))
       ktag = b'rsa'
+  if pub[b'k'] != b'rsa' and pub[b'k'] != b'ed25519':
+      raise KeyFormatError('unknown algorithm in k= tag: {0}'.format(pub[b'k']))
   return pk, keysize, ktag
 
 #: Abstract base class for holding messages and options during DKIM/ARC signing and verification.
diff --git a/dkim/tests/data/badk.txt b/dkim/tests/data/badk.txt
new file mode 100644
index 0000000..05d5580
--- /dev/null
+++ b/dkim/tests/data/badk.txt
@@ -0,0 +1 @@
+v=DKIM1; g=*; k=ed25519-sha256; p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDkHlOQoBTzWRiGs5V6NpP3idY6Wk08a5qhdR6wy5bdOKb2jLQiY/J16JYi0Qvx/byYzCNb3W91y3FutACDfzwQ/BC/e/8uBsCR+yz1Lxj+PL6lHvqMKrM3rG4hstT5QjvHO9PzoxZyVYLzBfO2EeC3Ip3G+2kryOTIKT+l/K4w3QIDAQAB
diff --git a/dkim/tests/test_dkim.py b/dkim/tests/test_dkim.py
index 12baf5f..c476daf 100644
--- a/dkim/tests/test_dkim.py
+++ b/dkim/tests/test_dkim.py
@@ -117,6 +117,30 @@ p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1Kd87/UeJjenpabgbFwh\
 s8g3FgD2Ap3ZB5DekAo5wMmk4wimDO+U8QzI3SD07y2+07wlNWwIt8svnxgdxGkVbb\
 hzY8i+RQ9DpSVpPbF7ykQxtKXkv/ahW3KjViiAH+ghvvIhkx4xYSIc9oSwVmAl5Oct\
 MEeWUwg8Istjqz8BZeTWbf41fbNhte7Y+YqZOwq1Sd0DbvYAD9NOZK9vlfuac0598H\
+Y+vtSBczUiKERHv1yRbcaQtZFh5wtiRrN04BLUTD21MycBX5jYchHjPY/wIDAQAB"""
+        }
+        try:
+            domain = domain.decode('ascii')
+        except UnicodeDecodeError:
+            return None
+        self.assertTrue(domain in _dns_responses,domain)
+        return _dns_responses[domain]
+
+    def dnsfunc4(self, domain):
+        sample_dns = """\
+k=rsa; \
+p=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBANmBe10IgY+u7h3enWTukkqtUD5PR52T\
+b/mPfjC0QJTocVBq6Za/PlzfV+Py92VaCak19F4WrbVTK5Gg5tW220MCAwEAAQ=="""
+
+        _dns_responses = {
+          'example._domainkey.canonical.com.': sample_dns,
+          'test._domainkey.example.com.': read_test_data("badk.txt"),
+          '20120113._domainkey.gmail.com.': """\
+p=MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1Kd87/UeJjenpabgbFwh\
++eBCsSTrqmwIYYvywlbhbqoo2DymndFkbjOVIPIldNs/m40KF+yzMn1skyoxcTUGCQ\
+s8g3FgD2Ap3ZB5DekAo5wMmk4wimDO+U8QzI3SD07y2+07wlNWwIt8svnxgdxGkVbb\
+hzY8i+RQ9DpSVpPbF7ykQxtKXkv/ahW3KjViiAH+ghvvIhkx4xYSIc9oSwVmAl5Oct\
+MEeWUwg8Istjqz8BZeTWbf41fbNhte7Y+YqZOwq1Sd0DbvYAD9NOZK9vlfuac0598H\
 Y+vtSBczUiKERHv1yRbcaQtZFh5wtiRrN04BLUTD21MycBX5jYchHjPY/wIDAQAB"""
         }
         try:
@@ -156,6 +180,16 @@ Y+vtSBczUiKERHv1yRbcaQtZFh5wtiRrN04BLUTD21MycBX5jYchHjPY/wIDAQAB"""
                 res = dkim.verify(sig + self.message, dnsfunc=self.dnsfunc3)
                 self.assertFalse(res)
 
+    def test_unknown_k(self):
+        # A error is detected if an unknown algorithm is in the k= tag.
+        for header_algo in (b"simple", b"relaxed"):
+            for body_algo in (b"simple", b"relaxed"):
+                sig = dkim.sign(
+                    self.message, b"test", b"example.com", self.key,
+                    canonicalize=(header_algo, body_algo))
+                res = dkim.verify(sig + self.message, dnsfunc=self.dnsfunc4)
+                self.assertFalse(res)
+
     def test_simple_signature(self):
         # A message verifies after being signed with SHOULD headers
         for header_algo in (b"simple", b"relaxed"):