| 75 | dns.tsig.validate(w, key, dns.name.from_text("foo.com"), tsig, now, b"", 0) |
| 76 | |
| 77 | def test_gssapi_context(self): |
| 78 | def verify_signature(data, mac): |
| 79 | if data == b"throw": |
| 80 | raise Exception |
| 81 | return None |
| 82 | |
| 83 | # mock out the gssapi context to return some dummy values |
| 84 | gssapi_context_mock = Mock() |
| 85 | gssapi_context_mock.get_signature.return_value = b"xxxxxxxxxxx" |
| 86 | gssapi_context_mock.verify_signature.side_effect = verify_signature |
| 87 | |
| 88 | # create the key and add it to the keyring |
| 89 | keyname = "gsstsigtest" |
| 90 | key = dns.tsig.Key(keyname, gssapi_context_mock, "gss-tsig") |
| 91 | ctx = dns.tsig.get_context(key) |
| 92 | self.assertEqual(ctx.name, "gss-tsig") |
| 93 | gsskeyname = dns.name.from_text(keyname) |
| 94 | keyring[gsskeyname] = key |
| 95 | |
| 96 | # make sure we can get the keyring (no exception == success) |
| 97 | text = dns.tsigkeyring.to_text(keyring) |
| 98 | self.assertNotEqual(text, "") |
| 99 | |
| 100 | # test exceptional case for _verify_mac_for_context |
| 101 | with self.assertRaises(dns.tsig.BadSignature): |
| 102 | ctx.update(b"throw") |
| 103 | ctx.verify(b"bogus") |
| 104 | gssapi_context_mock.verify_signature.assert_called() |
| 105 | self.assertEqual(gssapi_context_mock.verify_signature.call_count, 1) |
| 106 | |
| 107 | # simulate case where TKEY message is used to establish the context; |
| 108 | # first, the query from the client |
| 109 | tkey_message = dns.message.make_query(keyname, "tkey", "any") |
| 110 | |
| 111 | # test existent/non-existent keys in the keyring |
| 112 | adapted_keyring = dns.tsig.GSSTSigAdapter(keyring) |
| 113 | |
| 114 | fetched_key = adapted_keyring(tkey_message, gsskeyname) |
| 115 | self.assertEqual(fetched_key, key) |
| 116 | key = adapted_keyring(None, gsskeyname) |
| 117 | self.assertEqual(fetched_key, key) |
| 118 | key = adapted_keyring(tkey_message, "dummy") |
| 119 | self.assertEqual(key, None) |
| 120 | |
| 121 | # create a response, TKEY and turn it into bytes, simulating the server |
| 122 | # sending the response to the query |
| 123 | tkey_response = dns.message.make_response(tkey_message) |
| 124 | key = base64.b64decode("KEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEYKEY") |
| 125 | tkey = dns.rdtypes.ANY.TKEY.TKEY( |
| 126 | dns.rdataclass.ANY, |
| 127 | dns.rdatatype.TKEY, |
| 128 | dns.name.from_text("gss-tsig."), |
| 129 | 1594203795, |
| 130 | 1594206664, |
| 131 | 3, |
| 132 | 0, |
| 133 | key, |
| 134 | ) |