]>
Commit | Line | Data |
---|---|---|
1 | from __future__ import print_function | |
2 | import json | |
3 | import time | |
4 | import unittest | |
5 | from copy import deepcopy | |
6 | from pprint import pprint | |
7 | from test_helper import ApiTestCase, unique_tsigkey_name, is_auth, is_recursor, get_db_tsigkeys | |
8 | ||
9 | class AuthTSIGHelperMixin(object): | |
10 | def create_tsig_key(self, name=None, algorithm='hmac-md5', key=None): | |
11 | if name is None: | |
12 | name = unique_tsigkey_name() | |
13 | payload = { | |
14 | 'name': name, | |
15 | 'algorithm': algorithm, | |
16 | } | |
17 | if key is not None: | |
18 | payload.update({'key': key}) | |
19 | print("sending", payload) | |
20 | r = self.session.post( | |
21 | self.url("/api/v1/servers/localhost/tsigkeys"), | |
22 | data=json.dumps(payload), | |
23 | headers={'content-type': 'application/json'}) | |
24 | self.assert_success_json(r) | |
25 | self.assertEquals(r.status_code, 201) | |
26 | reply = r.json() | |
27 | print("reply", reply) | |
28 | return name, payload, reply | |
29 | ||
30 | ||
31 | @unittest.skipIf(not is_auth(), "Not applicable") | |
32 | class AuthTSIG(ApiTestCase, AuthTSIGHelperMixin): | |
33 | def test_create_key(self): | |
34 | """ | |
35 | Create a TSIG key that is generated by the server | |
36 | """ | |
37 | name, payload, data = self.create_tsig_key() | |
38 | for k in ('id', 'name', 'algorithm', 'key', 'type'): | |
39 | self.assertIn(k, data) | |
40 | if k in payload: | |
41 | self.assertEquals(data[k], payload[k]) | |
42 | ||
43 | def test_create_key_with_key_data(self): | |
44 | """ | |
45 | Create a new key with the key data provided | |
46 | """ | |
47 | key = 'fn+BREHMDq0uWA1WbDwaoc2ne3rD973ySJ33ToJTfWY=' | |
48 | name, payload, data = self.create_tsig_key(key=key) | |
49 | self.assertEqual(data['key'], key) | |
50 | ||
51 | def test_create_key_with_hmacsha512(self): | |
52 | """ | |
53 | Have the server generate a key with the provided algorithm | |
54 | """ | |
55 | algorithm = 'hmac-sha512' | |
56 | name, payload, data = self.create_tsig_key(algorithm=algorithm) | |
57 | self.assertEqual(data['algorithm'], algorithm) | |
58 | ||
59 | def test_get_non_existing_key(self): | |
60 | """ | |
61 | Try to get get a key that does not exist | |
62 | """ | |
63 | name = "idontexist" | |
64 | r = self.session.get(self.url( | |
65 | "/api/v1/servers/localhost/tsigkeys/" + name + '.'), | |
66 | headers={'accept': 'application/json'}) | |
67 | self.assert_error_json(r) | |
68 | self.assertEqual(r.status_code, 404) | |
69 | newdata = r.json() | |
70 | self.assertIn('TSIG key with name \'' + name + '\' not found', newdata['error']) | |
71 | ||
72 | def test_remove_key(self): | |
73 | """ | |
74 | Create a key and attempt to delete it | |
75 | """ | |
76 | name, payload, data = self.create_tsig_key() | |
77 | r = self.session.delete(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id'])) | |
78 | self.assertEqual(r.status_code, 204) | |
79 | keys_from_db = get_db_tsigkeys(name) | |
80 | self.assertListEqual(keys_from_db, []) | |
81 | ||
82 | def test_put_key_change_name(self): | |
83 | """ | |
84 | Rename a key by PUTing a json with "name" set | |
85 | """ | |
86 | name, payload, data = self.create_tsig_key() | |
87 | payload = { | |
88 | 'name': 'mynewkey' | |
89 | } | |
90 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), | |
91 | data=json.dumps(payload)) | |
92 | self.assertEqual(r.status_code, 200) | |
93 | newdata = r.json() | |
94 | self.assertEqual(newdata['name'], 'mynewkey') | |
95 | ||
96 | # Check if the old key is removed | |
97 | r = self.session.get(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id'])) | |
98 | self.assertEqual(r.status_code, 404, "Old key was not removed!") | |
99 | ||
100 | def test_put_key_change_key(self): | |
101 | """ | |
102 | Change the key by PUTing it | |
103 | """ | |
104 | name, payload, data = self.create_tsig_key() | |
105 | newkey = 'l36TAJalAys0HeEfSM1rFzSmz9kSwfiBo3HNkL62COs=' | |
106 | payload = { | |
107 | 'key': newkey | |
108 | } | |
109 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), | |
110 | data=json.dumps(payload)) | |
111 | self.assertEqual(r.status_code, 200) | |
112 | data = r.json() | |
113 | self.assertEqual(data['key'], newkey) | |
114 | ||
115 | def test_put_key_change_algo(self): | |
116 | name, payload, data = self.create_tsig_key() | |
117 | newalgo = 'hmac-sha256' | |
118 | payload = { | |
119 | 'algorithm': newalgo | |
120 | } | |
121 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), | |
122 | data=json.dumps(payload)) | |
123 | self.assertEqual(r.status_code, 200) | |
124 | data = r.json() | |
125 | self.assertEqual(data['algorithm'], newalgo) | |
126 | ||
127 | def test_put_non_existing_algo(self): | |
128 | name, payload, data = self.create_tsig_key() | |
129 | payload = { | |
130 | 'algorithm': 'foobar' | |
131 | } | |
132 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), | |
133 | data=json.dumps(payload)) | |
134 | self.assertEqual(r.status_code, 422) | |
135 | data = r.json() | |
136 | self.assertIn('Unknown TSIG algorithm: ', data['error']) | |
137 | ||
138 | def test_put_broken_key(self): | |
139 | name, payload, data = self.create_tsig_key() | |
140 | payload = { | |
141 | 'key': 'f\u0333oobar1======' | |
142 | } | |
143 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + data['id']), | |
144 | data=json.dumps(payload)) | |
145 | data = r.json() | |
146 | self.assertEqual(r.status_code, 422) | |
147 | self.assertIn('Can not base64 decode key content ', data['error']) | |
148 | ||
149 | def test_put_to_non_existing_key(self): | |
150 | name = unique_tsigkey_name() | |
151 | payload = { | |
152 | 'algorithm': 'hmac-sha512' | |
153 | } | |
154 | r = self.session.put(self.url("/api/v1/servers/localhost/tsigkeys/" + name + '.'), | |
155 | data=json.dumps(payload), | |
156 | headers={'accept': 'application/json'}) | |
157 | self.assertEqual(r.status_code, 404) | |
158 | data = r.json() | |
159 | self.assertIn('TSIG key with name \'' + name + '\' not found', data['error']) | |
160 | ||
161 | def test_post_existing_key_name(self): | |
162 | name, payload, data = self.create_tsig_key() | |
163 | r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), | |
164 | headers={'accept': 'application/json'}, | |
165 | data=json.dumps(payload)) | |
166 | self.assertEqual(r.status_code, 409) | |
167 | data = r.json() | |
168 | self.assertIn('A TSIG key with the name ', data['error']) | |
169 | ||
170 | def test_post_broken_key_name(self): | |
171 | payload = { | |
172 | 'name': unique_tsigkey_name(), | |
173 | 'key': 'f\u0333oobar1======', | |
174 | 'algorithm': 'hmac-md5' | |
175 | } | |
176 | r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), | |
177 | headers={'accept': 'application/json'}, | |
178 | data=json.dumps(payload)) | |
179 | self.assertEqual(r.status_code, 422) | |
180 | data = r.json() | |
181 | self.assertIn(' cannot be base64-decoded', data['error']) | |
182 | ||
183 | def test_post_wrong_algo(self): | |
184 | payload = { | |
185 | 'name': unique_tsigkey_name(), | |
186 | 'algorithm': 'foobar' | |
187 | } | |
188 | r = self.session.post(self.url("/api/v1/servers/localhost/tsigkeys"), | |
189 | headers={'accept': 'application/json'}, | |
190 | data=json.dumps(payload)) | |
191 | self.assertEqual(r.status_code, 400) | |
192 | data = r.json() | |
193 | self.assertIn('Invalid TSIG algorithm: ', data['error']) |