From: Pieter Lexis Date: Thu, 23 Nov 2017 11:15:00 +0000 (+0100) Subject: API: Add TSIG tests X-Git-Tag: auth-4.2.0-alpha1~17^2~21 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6eca65104de7316e1efbe613ff93cbcf3d12db85;p=thirdparty%2Fpdns.git API: Add TSIG tests --- diff --git a/regression-tests.api/test_TSIG.py b/regression-tests.api/test_TSIG.py new file mode 100644 index 0000000000..b677aa0755 --- /dev/null +++ b/regression-tests.api/test_TSIG.py @@ -0,0 +1,192 @@ +import json +import time +import unittest +from copy import deepcopy +from pprint import pprint +from test_helper import ApiTestCase, unique_tsigkey_name, is_auth, is_recursor, get_db_tsigkeys + +class AuthTSIGHelperMixin(object): + def create_tsig_key(self, name=None, algorithm='hmac-md5', key=None): + if name is None: + name = unique_tsigkey_name() + payload = { + 'name': name, + 'algorithm': algorithm, + } + if key is not None: + payload.update({'key': key}) + print "sending", payload + r = self.session.post( + self.url("/api/v1/servers/localhost/tsigkeys"), + data=json.dumps(payload), + headers={'content-type': 'application/json'}) + self.assert_success_json(r) + self.assertEquals(r.status_code, 201) + reply = r.json() + print "reply", reply + return name, payload, reply + + +@unittest.skipIf(not is_auth(), "Not applicable") +class AuthTSIG(ApiTestCase, AuthTSIGHelperMixin): + def test_create_key(self): + """ + Create a TSIG key that is generated by the server + """ + name, payload, data = self.create_tsig_key() + for k in ('id', 'name', 'algorithm', 'key', 'type'): + self.assertIn(k, data) + if k in payload: + self.assertEquals(data[k], payload[k]) + + def test_create_key_with_key_data(self): + """ + Create a new key with the key data provided + """ + key = 'fn+BREHMDq0uWA1WbDwaoc2ne3rD973ySJ33ToJTfWY=' + name, payload, data = self.create_tsig_key(key=key) + self.assertEqual(data['key'], key) + + def test_create_key_with_hmacsha512(self): + """ + Have the server generate a key with the provided algorithm + """ + algorithm = 'hmac-sha512' + name, payload, data = self.create_tsig_key(algorithm=algorithm) + self.assertEqual(data['algorithm'], algorithm) + + def test_get_non_existing_key(self): + """ + Try to get get a key that does not exist + """ + name = "idontexist" + r = self.session.get(self.url( + "/api/v1/servers/localhost/tsigkeys/" + name + '.'), + headers={'accept': 'application/json'}) + self.assert_error_json(r) + self.assertEqual(r.status_code, 404) + newdata = r.json() + self.assertIn('TSIG key with name \'' + name + '\' not found', newdata['error']) + + def test_remove_key(self): + """ + Create a key and attempt to delete it + """ + name, payload, data = self.create_tsig_key() + r = self.session.delete(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id'])) + self.assertEqual(r.status_code, 204) + keys_from_db = get_db_tsigkeys(name) + self.assertListEqual(keys_from_db, []) + + def test_put_key_change_name(self): + """ + Rename a key by PUTing a json with "name" set + """ + name, payload, data = self.create_tsig_key() + payload = { + 'name': 'mynewkey' + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), + data=json.dumps(payload)) + self.assertEqual(r.status_code, 200) + newdata = r.json() + self.assertEqual(newdata['name'], 'mynewkey') + + # Check if the old key is removed + r = self.session.get(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id'])) + self.assertEqual(r.status_code, 404, "Old key was not removed!") + + def test_put_key_change_key(self): + """ + Change the key by PUTing it + """ + name, payload, data = self.create_tsig_key() + newkey = 'l36TAJalAys0HeEfSM1rFzSmz9kSwfiBo3HNkL62COs=' + payload = { + 'key': newkey + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), + data=json.dumps(payload)) + self.assertEqual(r.status_code, 200) + data = r.json() + self.assertEqual(data['key'], newkey) + + def test_put_key_change_algo(self): + name, payload, data = self.create_tsig_key() + newalgo = 'hmac-sha256' + payload = { + 'algorithm': newalgo + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), + data=json.dumps(payload)) + self.assertEqual(r.status_code, 200) + data = r.json() + self.assertEqual(data['algorithm'], newalgo) + + def test_put_non_existing_algo(self): + name, payload, data = self.create_tsig_key() + payload = { + 'algorithm': 'foobar' + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), + data=json.dumps(payload)) + self.assertEqual(r.status_code, 422) + data = r.json() + self.assertIn('Unknown TSIG algorithm: ', data['error']) + + def test_put_broken_key(self): + name, payload, data = self.create_tsig_key() + payload = { + 'key': 'f\u333oobar1======' + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), + data=json.dumps(payload)) + data = r.json() + self.assertEqual(r.status_code, 422) + self.assertIn('Can not base64 decode key content ', data['error']) + + def test_put_to_non_existing_key(self): + name = unique_tsigkey_name() + payload = { + 'algorithm': 'hmac-sha512' + } + r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + name + '.'), + data=json.dumps(payload), + headers={'accept': 'application/json'}) + self.assertEqual(r.status_code, 404) + data = r.json() + self.assertIn('TSIG key with name \'' + name + '\' not found', data['error']) + + def test_post_existing_key_name(self): + name, payload, data = self.create_tsig_key() + r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), + headers={'accept': 'application/json'}, + data=json.dumps(payload)) + self.assertEqual(r.status_code, 422) + data = r.json() + self.assertIn('A TSIG key with the name ', data['error']) + + def test_post_broken_key_name(self): + payload = { + 'name': unique_tsigkey_name(), + 'key': 'f\u333oobar1======', + 'algorithm': 'hmac-md5' + } + r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), + headers={'accept': 'application/json'}, + data=json.dumps(payload)) + self.assertEqual(r.status_code, 422) + data = r.json() + self.assertIn(' cannot be base64-decoded', data['error']) + + def test_post_wrong_algo(self): + payload = { + 'name': unique_tsigkey_name(), + 'algorithm': 'foobar' + } + r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), + headers={'accept': 'application/json'}, + data=json.dumps(payload)) + self.assertEqual(r.status_code, 422) + data = r.json() + self.assertIn('Unknown TSIG algorithm: ', data['error']) diff --git a/regression-tests.api/test_helper.py b/regression-tests.api/test_helper.py index 2affa83a4b..bca9b0928f 100644 --- a/regression-tests.api/test_helper.py +++ b/regression-tests.api/test_helper.py @@ -53,6 +53,8 @@ class ApiTestCase(unittest.TestCase): def unique_zone_name(): return 'test-' + datetime.now().strftime('%d%H%S%M%f') + '.org.' +def unique_tsigkey_name(): + return 'test-' + datetime.now().strftime('%d%H%S%M%f') + '-key' def is_auth(): return DAEMON == 'authoritative' @@ -95,3 +97,14 @@ def sdig(*args): return subprocess.check_call([SDIG, '127.0.0.1', str(DNSPORT)] + list(args)) except subprocess.CalledProcessError as except_inst: raise RuntimeError("sdig %s %s failed: %s" % (command, args, except_inst.output.decode('ascii', errors='replace'))) + +def get_db_tsigkeys(keyname): + with get_auth_db() as db: + rows = db.execute(""" + SELECT name, algorithm, secret + FROM tsigkeys + WHERE name = ?""", (keyname, )).fetchall() + keys = [{'name': row[0], 'algorithm': row[1], 'secret': row[2]} for row in rows] + print "DB TSIG keys:", keys + return keys +