mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

OVH: Add support of DKIM records

Browse Source
This commit is contained in:
trnsnt
2017-10-23 17:12:32 +02:00
parent cf76cb1446
commit 6b1a8f8ccf
2 changed files with 231 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
octodns/provider/ovh.py
View File

@@ -5,6 +5,8 @@
from __future__ import absolute_import, division, print_function, \
unicode_literals
import base64
import binascii
import logging
from collections import defaultdict
@@ -32,8 +34,10 @@ class OvhProvider(BaseProvider):
SUPPORTS_GEO = False
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'MX', 'NAPTR', 'NS', 'PTR', 'SPF',
'SRV', 'SSHFP', 'TXT'))
# This variable is also used in populate method to filter which OVH record
# types are supported by octodns
SUPPORTS = set(('A', 'AAAA', 'CNAME', 'DKIM', 'MX', 'NAPTR', 'NS', 'PTR',
'SPF', 'SRV', 'SSHFP', 'TXT'))
def __init__(self, id, endpoint, application_key, application_secret,
consumer_key, *args, **kwargs):
@@ -62,6 +66,10 @@ class OvhProvider(BaseProvider):
before = len(zone.records)
for name, types in values.items():
for _type, records in types.items():
if _type not in self.SUPPORTS:
self.log.warning('Not managed record of type %s, skip',
_type)
continue
data_for = getattr(self, '_data_for_{}'.format(_type))
record = Record.new(zone, name, data_for(_type, records),
source=self, lenient=lenient)
@@ -96,7 +104,11 @@ class OvhProvider(BaseProvider):
def _apply_delete(self, zone_name, change):
existing = change.existing
self.delete_records(zone_name, existing._type, existing.name)
record_type = existing._type
if record_type == "TXT":
if self._is_valid_dkim(existing.values[0]):
record_type = 'DKIM'
self.delete_records(zone_name, record_type, existing.name)
@staticmethod
def _data_for_multiple(_type, records):
@@ -184,6 +196,15 @@ class OvhProvider(BaseProvider):
'values': values
}
@staticmethod
def _data_for_DKIM(_type, records):
return {
'ttl': records[0]['ttl'],
'type': "TXT",
'values': [record['target'].replace(';', '\;')
for record in records]
}
_data_for_A = _data_for_multiple
_data_for_AAAA = _data_for_multiple
_data_for_NS = _data_for_multiple
@@ -258,15 +279,63 @@ class OvhProvider(BaseProvider):
'fieldType': record._type
}
def _params_for_TXT(self, record):
for value in record.values:
field_type = 'TXT'
if self._is_valid_dkim(value):
field_type = 'DKIM'
value = value.replace("\;", ";")
yield {
'target': value,
'subDomain': record.name,
'ttl': record.ttl,
'fieldType': field_type
}
_params_for_A = _params_for_multiple
_params_for_AAAA = _params_for_multiple
_params_for_NS = _params_for_multiple
_params_for_SPF = _params_for_multiple
_params_for_TXT = _params_for_multiple
_params_for_CNAME = _params_for_single
_params_for_PTR = _params_for_single
def _is_valid_dkim(self, value):
"""Check if value is a valid DKIM"""
validator_dict = {'h': lambda val: val in ['sha1', 'sha256'],
's': lambda val: val in ['*', 'email'],
't': lambda val: val in ['y', 's'],
'v': lambda val: val == 'DKIM1',
'k': lambda val: val == 'rsa',
'n': lambda _: True,
'g': lambda _: True}
splitted = value.split('\;')
found_key = False
for splitted_value in splitted:
sub_split = map(lambda x: x.strip(), splitted_value.split("=", 1))
if len(sub_split) < 2:
return False
key, value = sub_split[0], sub_split[1]
if key == "p":
is_valid_key = self._is_valid_dkim_key(value)
if not is_valid_key:
return False
found_key = True
else:
is_valid_key = validator_dict.get(key, lambda _: False)(value)
if not is_valid_key:
return False
return found_key
@staticmethod
def _is_valid_dkim_key(key):
try:
base64.decodestring(key)
except binascii.Error:
return False
return True
def get_records(self, zone_name):
"""
List all records of a DNS zone

196
tests/test_octodns_provider_ovh.py
View File

@@ -17,6 +17,14 @@ from octodns.zone import Zone
class TestOvhProvider(TestCase):
api_record = []
valid_dkim = []
invalid_dkim = []
valid_dkim_key = "p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCxLaG16G4SaE" \
"cXVdiIxTg7gKSGbHKQLm30CHib1h9FzS9nkcyvQSyQj1rMFyqC//" \
"tft3ohx3nvJl+bGCWxdtLYDSmir9PW54e5CTdxEh8MWRkBO3StF6" \
"QG/tAh3aTGDmkqhIJGLb87iHvpmVKqURmEUzJPv5KPJfWLofADI+" \
"q9lQIDAQAB"
zone = Zone('unit.tests.', [])
expected = set()
@@ -233,6 +241,65 @@ class TestOvhProvider(TestCase):
'value': '1:1ec:1::1',
}))
# DKIM
api_record.append({
'fieldType': 'DKIM',
'ttl': 1300,
'target': valid_dkim_key,
'subDomain': 'dkim',
'id': 16
})
expected.add(Record.new(zone, 'dkim', {
'ttl': 1300,
'type': 'TXT',
'value': valid_dkim_key,
}))
# TXT
api_record.append({
'fieldType': 'TXT',
'ttl': 1400,
'target': 'TXT text',
'subDomain': 'txt',
'id': 17
})
expected.add(Record.new(zone, 'txt', {
'ttl': 1400,
'type': 'TXT',
'value': 'TXT text',
}))
# LOC
# We do not have associated record for LOC, as it's not managed
api_record.append({
'fieldType': 'LOC',
'ttl': 1500,
'target': '1 1 1 N 1 1 1 E 1m 1m',
'subDomain': '',
'id': 18
})
valid_dkim = [valid_dkim_key,
'v=DKIM1 \; %s' % valid_dkim_key,
'h=sha256 \; %s' % valid_dkim_key,
'h=sha1 \; %s' % valid_dkim_key,
's=* \; %s' % valid_dkim_key,
's=email \; %s' % valid_dkim_key,
't=y \; %s' % valid_dkim_key,
't=s \; %s' % valid_dkim_key,
'k=rsa \; %s' % valid_dkim_key,
'n=notes \; %s' % valid_dkim_key,
'g=granularity \; %s' % valid_dkim_key,
]
invalid_dkim = ['p=%invalid%', # Invalid public key
'v=DKIM1', # Missing public key
'v=DKIM2 \; %s' % valid_dkim_key, # Invalid version
'h=sha512 \; %s' % valid_dkim_key, # Invalid hash algo
's=fake \; %s' % valid_dkim_key, # Invalid selector
't=fake \; %s' % valid_dkim_key, # Invalid flag
'u=invalid \; %s' % valid_dkim_key, # Invalid key
]
@patch('ovh.Client')
def test_populate(self, client_mock):
provider = OvhProvider('test', 'endpoint', 'application_key',
@@ -253,6 +320,16 @@ class TestOvhProvider(TestCase):
provider.populate(zone)
self.assertEquals(self.expected, zone.records)
@patch('ovh.Client')
def test_is_valid_dkim(self, client_mock):
"""Test _is_valid_dkim"""
provider = OvhProvider('test', 'endpoint', 'application_key',
'application_secret', 'consumer_key')
for dkim in self.valid_dkim:
self.assertTrue(provider._is_valid_dkim(dkim))
for dkim in self.invalid_dkim:
self.assertFalse(provider._is_valid_dkim(dkim))
@patch('ovh.Client')
def test_apply(self, client_mock):
provider = OvhProvider('test', 'endpoint', 'application_key',
@@ -270,90 +347,91 @@ class TestOvhProvider(TestCase):
provider.apply(plan)
self.assertEquals(get_mock.side_effect, ctx.exception)
# Records get by API call
with patch.object(provider._client, 'get') as get_mock:
get_returns = [[1, 2], {
'fieldType': 'A',
'ttl': 600,
'target': '5.6.7.8',
'subDomain': '',
'id': 100
}, {'fieldType': 'A',
'ttl': 600,
'target': '5.6.7.8',
'subDomain': 'fake',
'id': 101
}]
get_returns = [
[1, 2, 3, 4],
{'fieldType': 'A', 'ttl': 600, 'target': '5.6.7.8',
'subDomain': '', 'id': 100},
{'fieldType': 'A', 'ttl': 600, 'target': '5.6.7.8',
'subDomain': 'fake', 'id': 101},
{'fieldType': 'TXT', 'ttl': 600, 'target': 'fake txt record',
'subDomain': 'txt', 'id': 102},
{'fieldType': 'DKIM', 'ttl': 600,
'target': 'v=DKIM1; %s' % self.valid_dkim_key,
'subDomain': 'dkim', 'id': 103}
]
get_mock.side_effect = get_returns
plan = provider.plan(desired)
with patch.object(provider._client, 'post') as post_mock:
with patch.object(provider._client, 'delete') as delete_mock:
with patch.object(provider._client, 'get') as get_mock:
get_mock.side_effect = [[100], [101]]
with patch.object(provider._client, 'post') as post_mock, \
patch.object(provider._client, 'delete') as delete_mock:
get_mock.side_effect = [[100], [101], [102], [103]]
provider.apply(plan)
wanted_calls = [
call(u'/domain/zone/unit.tests/record',
fieldType=u'A',
call(u'/domain/zone/unit.tests/record', fieldType=u'TXT',
subDomain='txt', target=u'TXT text', ttl=1400),
call(u'/domain/zone/unit.tests/record', fieldType=u'DKIM',
subDomain='dkim', target=self.valid_dkim_key,
ttl=1300),
call(u'/domain/zone/unit.tests/record', fieldType=u'A',
subDomain=u'', target=u'1.2.3.4', ttl=100),
call(u'/domain/zone/unit.tests/record',
fieldType=u'SRV',
call(u'/domain/zone/unit.tests/record', fieldType=u'SRV',
subDomain=u'10 20 30 foo-1.unit.tests.',
target='_srv._tcp', ttl=800),
call(u'/domain/zone/unit.tests/record',
fieldType=u'SRV',
call(u'/domain/zone/unit.tests/record', fieldType=u'SRV',
subDomain=u'40 50 60 foo-2.unit.tests.',
target='_srv._tcp', ttl=800),
call(u'/domain/zone/unit.tests/record', fieldType=u'PTR',
subDomain='4', target=u'unit.tests.', ttl=900),
call(u'/domain/zone/unit.tests/record', fieldType=u'NS',
subDomain='www3', target=u'ns3.unit.tests.', ttl=700),
call(u'/domain/zone/unit.tests/record', fieldType=u'NS',
subDomain='www3', target=u'ns4.unit.tests.', ttl=700),
call(u'/domain/zone/unit.tests/record',
fieldType=u'PTR', subDomain='4',
target=u'unit.tests.', ttl=900),
call(u'/domain/zone/unit.tests/record',
fieldType=u'NS', subDomain='www3',
target=u'ns3.unit.tests.', ttl=700),
call(u'/domain/zone/unit.tests/record',
fieldType=u'NS', subDomain='www3',
target=u'ns4.unit.tests.', ttl=700),
call(u'/domain/zone/unit.tests/record',
fieldType=u'SSHFP',
fieldType=u'SSHFP', target=u'', ttl=1100,
subDomain=u'1 1 bf6b6825d2977c511a475bbefb88a'
u'ad54'
u'a92ac73',
target=u'', ttl=1100),
call(u'/domain/zone/unit.tests/record',
fieldType=u'AAAA', subDomain=u'',
target=u'1:1ec:1::1', ttl=200),
call(u'/domain/zone/unit.tests/record',
fieldType=u'MX', subDomain=u'',
target=u'10 mx1.unit.tests.', ttl=400),
call(u'/domain/zone/unit.tests/record',
fieldType=u'CNAME', subDomain='www2',
target=u'unit.tests.', ttl=300),
call(u'/domain/zone/unit.tests/record',
fieldType=u'SPF', subDomain=u'',
),
call(u'/domain/zone/unit.tests/record', fieldType=u'AAAA',
subDomain=u'', target=u'1:1ec:1::1', ttl=200),
call(u'/domain/zone/unit.tests/record', fieldType=u'MX',
subDomain=u'', target=u'10 mx1.unit.tests.', ttl=400),
call(u'/domain/zone/unit.tests/record', fieldType=u'CNAME',
subDomain='www2', target=u'unit.tests.', ttl=300),
call(u'/domain/zone/unit.tests/record', fieldType=u'SPF',
subDomain=u'', ttl=1000,
target=u'v=spf1 include:unit.texts.'
u'rerirect ~all',
ttl=1000),
call(u'/domain/zone/unit.tests/record',
fieldType=u'A',
),
call(u'/domain/zone/unit.tests/record', fieldType=u'A',
subDomain='sub', target=u'1.2.3.4', ttl=200),
call(u'/domain/zone/unit.tests/record',
fieldType=u'NAPTR', subDomain='naptr',
call(u'/domain/zone/unit.tests/record', fieldType=u'NAPTR',
subDomain='naptr', ttl=500,
target=u'10 100 "S" "SIP+D2U" "!^.*$!sip:'
u'info@bar'
u'.example.com!" .',
ttl=500),
u'.example.com!" .'
),
call(u'/domain/zone/unit.tests/refresh')]
post_mock.assert_has_calls(wanted_calls)
# Get for delete calls
get_mock.assert_has_calls(
[call(u'/domain/zone/unit.tests/record',
fieldType=u'A', subDomain=u''),
call(u'/domain/zone/unit.tests/record',
fieldType=u'A', subDomain='fake')]
)
# 2 delete calls, one for update + one for delete
wanted_get_calls = [
call(u'/domain/zone/unit.tests/record', fieldType=u'TXT',
subDomain='txt'),
call(u'/domain/zone/unit.tests/record', fieldType=u'DKIM',
subDomain='dkim'),
call(u'/domain/zone/unit.tests/record', fieldType=u'A',
subDomain=u''),
call(u'/domain/zone/unit.tests/record', fieldType=u'A',
subDomain='fake')]
get_mock.assert_has_calls(wanted_get_calls)
# 4 delete calls for update and delete
delete_mock.assert_has_calls(
[call(u'/domain/zone/unit.tests/record/100'),
call(u'/domain/zone/unit.tests/record/101')])
call(u'/domain/zone/unit.tests/record/101'),
call(u'/domain/zone/unit.tests/record/102'),
call(u'/domain/zone/unit.tests/record/103')])