Merge pull request #530 from nrhall/nrhall-gss-tsig-changes

Add support for gss-tsig and TKEY records to support GSSAPI authentication

2 files changed, 246 insertions, 1 deletions

diff --git a/tests/test_rdtypeanytkey.py b/tests/test_rdtypeanytkey.py
new file mode 100644
index 0000000..3a3ca57
--- /dev/null
+++ b/tests/test_rdtypeanytkey.py
@@ -0,0 +1,96 @@
+# -*- coding: utf-8
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-2007, 2009-2011 Nominum, Inc.
+#
+# Permission to use, copy, modify, and distribute this software and its
+# documentation for any purpose with or without fee is hereby granted,
+# provided that the above copyright notice and this permission notice
+# appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
+# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+import unittest
+import base64
+
+import dns.name
+import dns.zone
+import dns.rdtypes.ANY.TKEY
+from dns.rdataclass import RdataClass
+from dns.rdatatype import RdataType
+
+
+class RdtypeAnyTKeyTestCase(unittest.TestCase):
+    tkey_rdata_text = 'gss-tsig. 1594203795 1594206664 3 0 KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY OTHEROTHEROTHEROTHEROTHEROTHEROT'
+    tkey_rdata_text_no_other = 'gss-tsig. 1594203795 1594206664 3 0 KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY'
+
+    def testTextOptionalData(self):
+        # construct the rdata from text and extract the TKEY
+        tkey = dns.rdata.from_text(
+            RdataClass.ANY, RdataType.TKEY,
+            RdtypeAnyTKeyTestCase.tkey_rdata_text, origin='.')
+        self.assertEqual(type(tkey), dns.rdtypes.ANY.TKEY.TKEY)
+
+        # go to text and compare
+        tkey_out_text = tkey.to_text(relativize=False)
+        self.assertEqual(tkey_out_text,
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text)
+
+    def testTextNoOptionalData(self):
+        # construct the rdata from text and extract the TKEY
+        tkey = dns.rdata.from_text(
+            RdataClass.ANY, RdataType.TKEY,
+            RdtypeAnyTKeyTestCase.tkey_rdata_text_no_other, origin='.')
+        self.assertEqual(type(tkey), dns.rdtypes.ANY.TKEY.TKEY)
+
+        # go to text and compare
+        tkey_out_text = tkey.to_text(relativize=False)
+        self.assertEqual(tkey_out_text,
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text_no_other)
+
+    def testWireOptionalData(self):
+        key = base64.b64decode('KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY')
+        other = base64.b64decode('OTHEROTHEROTHEROTHEROTHEROTHEROT')
+
+        # construct the TKEY and compare the text output
+        tkey = dns.rdtypes.ANY.TKEY.TKEY(dns.rdataclass.ANY,
+                                         dns.rdatatype.TKEY,
+                                         dns.name.from_text('gss-tsig.'),
+                                         1594203795, 1594206664,
+                                         3, 0, key, other)
+        self.assertEqual(tkey.to_text(relativize=False),
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text)
+
+        # go to/from wire and compare the text output
+        wire = tkey.to_wire()
+        tkey_out_wire = dns.rdata.from_wire(dns.rdataclass.ANY,
+                                            dns.rdatatype.TKEY,
+                                            wire, 0, len(wire))
+        self.assertEqual(tkey_out_wire.to_text(relativize=False),
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text)
+
+    def testWireNoOptionalData(self):
+        key = base64.b64decode('KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY')
+
+        # construct the TKEY with no 'other' data and compare the text output
+        tkey = dns.rdtypes.ANY.TKEY.TKEY(dns.rdataclass.ANY,
+                                         dns.rdatatype.TKEY,
+                                         dns.name.from_text('gss-tsig.'),
+                                         1594203795, 1594206664,
+                                         3, 0, key)
+        self.assertEqual(tkey.to_text(relativize=False),
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text_no_other)
+
+        # go to/from wire and compare the text output
+        wire = tkey.to_wire()
+        tkey_out_wire = dns.rdata.from_wire(dns.rdataclass.ANY,
+                                            dns.rdatatype.TKEY,
+                                            wire, 0, len(wire))
+        self.assertEqual(tkey_out_wire.to_text(relativize=False),
+                         RdtypeAnyTKeyTestCase.tkey_rdata_text_no_other)
diff --git a/tests/test_tsig.py b/tests/test_tsig.py
index f97e53b..ec5a6cc 100644
--- a/tests/test_tsig.py
+++ b/tests/test_tsig.py
@@ -1,13 +1,15 @@
 # Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
 
-import hashlib
 import unittest
+from unittest.mock import Mock
 import time
+import base64
 
 import dns.rcode
 import dns.tsig
 import dns.tsigkeyring
 import dns.message
+import dns.rdtypes.ANY.TKEY
 
 keyring = dns.tsigkeyring.from_text(
     {
@@ -17,6 +19,7 @@ keyring = dns.tsigkeyring.from_text(
 
 keyname = dns.name.from_text('keyname')
 
+
 class TSIGTestCase(unittest.TestCase):
 
     def test_get_context(self):
@@ -43,6 +46,130 @@ class TSIGTestCase(unittest.TestCase):
         m.use_tsig(keyring, keyname, tsig_error=dns.rcode.BADKEY)
         self.assertEqual(m.tsig_error, dns.rcode.BADKEY)
 
+    def test_verify_mac_for_context(self):
+        key = dns.tsig.Key('foo.com', 'abcd', 'hmac-sha512')
+        ctx = dns.tsig.get_context(key)
+        bad_expected = b'xxxxxxxxxx'
+        with self.assertRaises(dns.tsig.BadSignature):
+            ctx.verify(bad_expected)
+
+    def test_validate(self):
+        # make message and grab the TSIG
+        m = dns.message.make_query('example', 'a')
+        m.use_tsig(keyring, keyname, algorithm=dns.tsig.HMAC_SHA256)
+        w = m.to_wire()
+        tsig = m.tsig[0]
+
+        # get the time and create a key with matching characteristics
+        now = int(time.time())
+        key = dns.tsig.Key('foo.com', 'abcd', 'hmac-sha256')
+
+        # add enough to the time to take it over the fudge amount
+        with self.assertRaises(dns.tsig.BadTime):
+            dns.tsig.validate(w, key, dns.name.from_text('foo.com'),
+                              tsig, now + 1000, b'', 0)
+
+        # change the key name
+        with self.assertRaises(dns.tsig.BadKey):
+            dns.tsig.validate(w, key, dns.name.from_text('bar.com'),
+                              tsig, now, b'', 0)
+
+        # change the key algorithm
+        key = dns.tsig.Key('foo.com', 'abcd', 'hmac-sha512')
+        with self.assertRaises(dns.tsig.BadAlgorithm):
+            dns.tsig.validate(w, key, dns.name.from_text('foo.com'),
+                              tsig, now, b'', 0)
+
+    def test_gssapi_context(self):
+        def verify_signature(data, mac):
+            if data == b'throw':
+                raise Exception
+            return None
+
+        # mock out the gssapi context to return some dummy values
+        gssapi_context_mock = Mock()
+        gssapi_context_mock.get_signature.return_value = b'xxxxxxxxxxx'
+        gssapi_context_mock.verify_signature.side_effect = verify_signature
+
+        # create the key and add it to the keyring
+        keyname = 'gsstsigtest'
+        key = dns.tsig.Key(keyname, gssapi_context_mock, 'gss-tsig')
+        ctx = dns.tsig.get_context(key)
+        self.assertEqual(ctx.name, 'gss-tsig')
+        gsskeyname = dns.name.from_text(keyname)
+        keyring[gsskeyname] = key
+
+        # make sure we can get the keyring (no exception == success)
+        text = dns.tsigkeyring.to_text(keyring)
+        self.assertNotEqual(text, '')
+
+        # test exceptional case for _verify_mac_for_context
+        with self.assertRaises(dns.tsig.BadSignature):
+            ctx.update(b'throw')
+            ctx.verify(b'bogus')
+        gssapi_context_mock.verify_signature.assert_called()
+        self.assertEqual(gssapi_context_mock.verify_signature.call_count, 1)
+
+        # simulate case where TKEY message is used to establish the context;
+        # first, the query from the client
+        tkey_message = dns.message.make_query(keyname, 'tkey', 'any')
+
+        # test existent/non-existent keys in the keyring
+        adapted_keyring = dns.tsig.GSSTSigAdapter(keyring)
+
+        fetched_key = adapted_keyring(tkey_message, gsskeyname)
+        self.assertEqual(fetched_key, key)
+        key = adapted_keyring(None, gsskeyname)
+        self.assertEqual(fetched_key, key)
+        key = adapted_keyring(tkey_message, "dummy")
+        self.assertEqual(key, None)
+
+        # create a response, TKEY and turn it into bytes, simulating the server
+        # sending the response to the query
+        tkey_response = dns.message.make_response(tkey_message)
+        key = base64.b64decode('KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY')
+        tkey = dns.rdtypes.ANY.TKEY.TKEY(dns.rdataclass.ANY,
+                                         dns.rdatatype.TKEY,
+                                         dns.name.from_text('gss-tsig.'),
+                                         1594203795, 1594206664,
+                                         3, 0, key)
+
+        # add the TKEY answer and sign it
+        tkey_response.set_rcode(dns.rcode.NOERROR)
+        tkey_response.answer = [
+            dns.rrset.from_rdata(dns.name.from_text(keyname), 0, tkey)]
+        tkey_response.use_tsig(keyring=dns.tsig.GSSTSigAdapter(keyring),
+                               keyname=gsskeyname,
+                               algorithm=dns.tsig.GSS_TSIG)
+
+        # "send" it to the client
+        tkey_wire = tkey_response.to_wire()
+
+        # grab the response from the "server" and simulate the client side
+        dns.message.from_wire(tkey_wire, dns.tsig.GSSTSigAdapter(keyring))
+
+        # assertions to make sure the "gssapi" functions were called
+        gssapi_context_mock.get_signature.assert_called()
+        self.assertEqual(gssapi_context_mock.get_signature.call_count, 1)
+        gssapi_context_mock.verify_signature.assert_called()
+        self.assertEqual(gssapi_context_mock.verify_signature.call_count, 2)
+        gssapi_context_mock.step.assert_called()
+        self.assertEqual(gssapi_context_mock.step.call_count, 1)
+
+        # create example message and go to/from wire to simulate sign/verify
+        # of regular messages
+        a_message = dns.message.make_query('example', 'a')
+        a_message.use_tsig(dns.tsig.GSSTSigAdapter(keyring), gsskeyname)
+        a_wire = a_message.to_wire()
+        # not raising is passing
+        dns.message.from_wire(a_wire, dns.tsig.GSSTSigAdapter(keyring))
+
+        # assertions to make sure the "gssapi" functions were called again
+        gssapi_context_mock.get_signature.assert_called()
+        self.assertEqual(gssapi_context_mock.get_signature.call_count, 2)
+        gssapi_context_mock.verify_signature.assert_called()
+        self.assertEqual(gssapi_context_mock.verify_signature.call_count, 3)
+
     def test_sign_and_validate(self):
         m = dns.message.make_query('example', 'a')
         m.use_tsig(keyring, keyname)
@@ -50,6 +177,18 @@ class TSIGTestCase(unittest.TestCase):
         # not raising is passing
         dns.message.from_wire(w, keyring)
 
+    def test_validate_with_bad_keyring(self):
+        m = dns.message.make_query('example', 'a')
+        m.use_tsig(keyring, keyname)
+        w = m.to_wire()
+
+        # keyring == None is an error
+        with self.assertRaises(dns.message.UnknownTSIGKey):
+            dns.message.from_wire(w, None)
+        # callable keyring that returns None is an error
+        with self.assertRaises(dns.message.UnknownTSIGKey):
+            dns.message.from_wire(w, lambda m, n: None)
+
     def test_sign_and_validate_with_other_data(self):
         m = dns.message.make_query('example', 'a')
         m.use_tsig(keyring, keyname, other_data=b'other')
@@ -57,6 +196,16 @@ class TSIGTestCase(unittest.TestCase):
         # not raising is passing
         dns.message.from_wire(w, keyring)
 
+    def test_sign_respond_and_validate(self):
+        mq = dns.message.make_query('example', 'a')
+        mq.use_tsig(keyring, keyname)
+        wq = mq.to_wire()
+        mq_with_tsig = dns.message.from_wire(wq, keyring)
+        mr = dns.message.make_response(mq)
+        mr.use_tsig(keyring, keyname)
+        wr = mr.to_wire()
+        dns.message.from_wire(wr, keyring, request_mac=mq_with_tsig.mac)
+
     def make_message_pair(self, qname='example', rdtype='A', tsig_error=0):
         q = dns.message.make_query(qname, rdtype)
         q.use_tsig(keyring=keyring, keyname=keyname)