forked from octodns/octodns
-
Notifications
You must be signed in to change notification settings - Fork 0
/
selectel.py
305 lines (249 loc) · 9.86 KB
/
selectel.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
#
#
#
from __future__ import absolute_import, division, print_function, \
unicode_literals
from collections import defaultdict
from logging import getLogger
from requests import Session
from ..record import Record, Update
from .base import BaseProvider
class SelectelAuthenticationRequired(Exception):
def __init__(self, msg):
message = 'Authorization failed. Invalid or empty token.'
super(SelectelAuthenticationRequired, self).__init__(message)
class SelectelProvider(BaseProvider):
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NS', 'TXT', 'SPF', 'SRV'))
MIN_TTL = 60
PAGINATION_LIMIT = 50
API_URL = 'https://api.selectel.ru/domains/v1'
def __init__(self, id, token, *args, **kwargs):
self.log = getLogger('SelectelProvider[{}]'.format(id))
self.log.debug('__init__: id=%s', id)
super(SelectelProvider, self).__init__(id, *args, **kwargs)
self._sess = Session()
self._sess.headers.update({
'X-Token': token,
'Content-Type': 'application/json',
})
self._zone_records = {}
self._domain_list = self.domain_list()
self._zones = None
def _request(self, method, path, params=None, data=None):
self.log.debug('_request: method=%s, path=%s', method, path)
url = '{}{}'.format(self.API_URL, path)
resp = self._sess.request(method, url, params=params, json=data)
self.log.debug('_request: status=%s', resp.status_code)
if resp.status_code == 401:
raise SelectelAuthenticationRequired(resp.text)
elif resp.status_code == 404:
return {}
resp.raise_for_status()
if method == 'DELETE':
return {}
return resp.json()
def _get_total_count(self, path):
url = '{}{}'.format(self.API_URL, path)
resp = self._sess.request('HEAD', url)
return int(resp.headers['X-Total-Count'])
def _request_with_pagination(self, path, total_count):
result = []
for offset in range(0, total_count, self.PAGINATION_LIMIT):
result += self._request('GET', path,
params={'limit': self.PAGINATION_LIMIT,
'offset': offset})
return result
def _include_change(self, change):
if isinstance(change, Update):
existing = change.existing.data
new = change.new.data
new['ttl'] = max(self.MIN_TTL, new['ttl'])
if new == existing:
self.log.debug('_include_changes: new=%s, found existing=%s',
new, existing)
return False
return True
def _apply(self, plan):
desired = plan.desired
changes = plan.changes
self.log.debug('_apply: zone=%s, len(changes)=%d', desired.name,
len(changes))
zone_name = desired.name[:-1]
for change in changes:
class_name = change.__class__.__name__
getattr(self, '_apply_{}'.format(class_name).lower())(zone_name,
change)
def _apply_create(self, zone_name, change):
new = change.new
params_for = getattr(self, '_params_for_{}'.format(new._type))
for params in params_for(new):
self.create_record(zone_name, params)
def _apply_update(self, zone_name, change):
self._apply_delete(zone_name, change)
self._apply_create(zone_name, change)
def _apply_delete(self, zone_name, change):
existing = change.existing
self.delete_record(zone_name, existing._type, existing.name)
def _params_for_multiple(self, record):
for value in record.values:
yield {
'content': value,
'name': record.fqdn,
'ttl': max(self.MIN_TTL, record.ttl),
'type': record._type,
}
def _params_for_single(self, record):
yield {
'content': record.value,
'name': record.fqdn,
'ttl': max(self.MIN_TTL, record.ttl),
'type': record._type
}
def _params_for_MX(self, record):
for value in record.values:
yield {
'content': value.exchange,
'name': record.fqdn,
'ttl': max(self.MIN_TTL, record.ttl),
'type': record._type,
'priority': value.preference
}
def _params_for_SRV(self, record):
for value in record.values:
yield {
'name': record.fqdn,
'target': value.target,
'ttl': max(self.MIN_TTL, record.ttl),
'type': record._type,
'port': value.port,
'weight': value.weight,
'priority': value.priority
}
_params_for_A = _params_for_multiple
_params_for_AAAA = _params_for_multiple
_params_for_NS = _params_for_multiple
_params_for_TXT = _params_for_multiple
_params_for_SPF = _params_for_multiple
_params_for_CNAME = _params_for_single
def _data_for_A(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': [r['content'] for r in records],
}
_data_for_AAAA = _data_for_A
def _data_for_NS(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': ['{}.'.format(r['content']) for r in records],
}
def _data_for_MX(self, _type, records):
values = []
for record in records:
values.append({
'preference': record['priority'],
'exchange': '{}.'.format(record['content']),
})
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': values,
}
def _data_for_CNAME(self, _type, records):
only = records[0]
return {
'ttl': only['ttl'],
'type': _type,
'value': '{}.'.format(only['content'])
}
def _data_for_TXT(self, _type, records):
return {
'ttl': records[0]['ttl'],
'type': _type,
'values': [r['content'] for r in records],
}
def _data_for_SRV(self, _type, records):
values = []
for record in records:
values.append({
'priority': record['priority'],
'weight': record['weight'],
'port': record['port'],
'target': '{}.'.format(record['target']),
})
return {
'type': _type,
'ttl': records[0]['ttl'],
'values': values,
}
def populate(self, zone, target=False, lenient=False):
self.log.debug('populate: name=%s, target=%s, lenient=%s',
zone.name, target, lenient)
before = len(zone.records)
records = self.zone_records(zone)
if records:
values = defaultdict(lambda: defaultdict(list))
for record in records:
name = zone.hostname_from_fqdn(record['name'])
_type = record['type']
if _type in self.SUPPORTS:
values[name][record['type']].append(record)
for name, types in values.items():
for _type, records in types.items():
data_for = getattr(self, '_data_for_{}'.format(_type))
data = data_for(_type, records)
record = Record.new(zone, name, data, source=self,
lenient=lenient)
zone.add_record(record)
self.log.info('populate: found %s records',
len(zone.records) - before)
def domain_list(self):
path = '/'
domains = {}
domains_list = []
total_count = self._get_total_count(path)
domains_list = self._request_with_pagination(path, total_count)
for domain in domains_list:
domains[domain['name']] = domain
return domains
def zone_records(self, zone):
path = '/{}/records/'.format(zone.name[:-1])
zone_records = []
total_count = self._get_total_count(path)
zone_records = self._request_with_pagination(path, total_count)
self._zone_records[zone.name] = zone_records
return self._zone_records[zone.name]
def create_domain(self, name, zone=""):
path = '/'
data = {
'name': name,
'bind_zone': zone,
}
resp = self._request('POST', path, data=data)
self._domain_list[name] = resp
return resp
def create_record(self, zone_name, data):
self.log.debug('Create record. Zone: %s, data %s', zone_name, data)
if zone_name in self._domain_list.keys():
domain_id = self._domain_list[zone_name]['id']
else:
domain_id = self.create_domain(zone_name)['id']
path = '/{}/records/'.format(domain_id)
return self._request('POST', path, data=data)
def delete_record(self, domain, _type, zone):
self.log.debug('Delete record. Domain: %s, Type: %s', domain, _type)
domain_id = self._domain_list[domain]['id']
records = self._zone_records.get('{}.'.format(domain), False)
if not records:
path = '/{}/records/'.format(domain_id)
records = self._request('GET', path)
for record in records:
full_domain = domain
if zone:
full_domain = '{}{}'.format(zone, domain)
if record['type'] == _type and record['name'] == full_domain:
path = '/{}/records/{}'.format(domain_id, record['id'])
return self._request('DELETE', path)
self.log.debug('Delete record failed (Record not found)')