openstack/designate
Code Issues Proposed changes

Merge "CERT DNS records"

Browse Source
This commit is contained in:
Zuul 2021-08-31 00:49:33 +00:00 committed by Gerrit Code Review
commit 76bb79dd0d
11 changed files with 261 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
contrib/archive/backends/impl_ipa/__init__.py
View File

@ -65,6 +65,7 @@ rectype2iparectype = {'A': ('arecord', '%(data)s'),
'SSHFP': ('sshfprecord', '%(data)s'),
'NAPTR': ('naptrrecord', '%(data)s'),
'CAA': ('caarecord', '%(data)s'),
'CERT': ('certrecord', '%(data)s'),
}

2
designate/conf/base.py
View File

@ -54,7 +54,7 @@ DESIGNATE_OPTS = [
# Supported record types
cfg.ListOpt('supported_record_type', help='Supported record types',
default=['A', 'AAAA', 'CNAME', 'MX', 'SRV', 'TXT', 'SPF', 'NS',
'PTR', 'SSHFP', 'SOA', 'NAPTR', 'CAA']),
'PTR', 'SSHFP', 'SOA', 'NAPTR', 'CAA', 'CERT']),
# TCP Settings
cfg.IntOpt('backlog',

1
designate/objects/__init__.py
View File

@ -49,6 +49,7 @@ from designate.objects.zone_export import ZoneExport, ZoneExportList # noqa
from designate.objects.rrdata_a import A, AList # noqa
from designate.objects.rrdata_aaaa import AAAA, AAAAList # noqa
from designate.objects.rrdata_caa import CAA, CAAList # noqa
from designate.objects.rrdata_cert import CERT, CERTList # noqa
from designate.objects.rrdata_cname import CNAME, CNAMEList # noqa
from designate.objects.rrdata_mx import MX, MXList # noqa
from designate.objects.rrdata_naptr import NAPTR, NAPTRList # noqa

26
designate/objects/fields.py
View File

@ -102,6 +102,8 @@ class StringFields(ovoo_fields.StringField):
RE_KVP = r'^\s[A-Za-z0-9]+=[A-Za-z0-9]+'
RE_URL_MAIL = r'^mailto:[A-Za-z0-9_\-]+@.*'
RE_URL_HTTP = r'^http(s)?://.*/'
RE_CERT_TYPE = r'(^[A-Z]+$)|(^[0-9]+$)'
RE_CERT_ALGO = r'(^[A-Z]+[A-Z0-9\-]+[A-Z0-9]$)|(^[0-9]+$)'
def __init__(self, nullable=False, read_only=False,
default=ovoo_fields.UnspecifiedDefault, description='',
@ -389,6 +391,30 @@ class CaaPropertyField(StringFields):
return value
class CertTypeField(StringFields):
def __init__(self, **kwargs):
super(CertTypeField, self).__init__(**kwargs)
def coerce(self, obj, attr, value):
value = super(CertTypeField, self).coerce(obj, attr, value)
if not re.match(self.RE_CERT_TYPE, "%s" % value):
raise ValueError("Cert type %s is not a valid Mnemonic or "
"value" % value)
return value
class CertAlgoField(StringFields):
def __init__(self, **kwargs):
super(CertAlgoField, self).__init__(**kwargs)
def coerce(self, obj, attr, value):
value = super(CertAlgoField, self).coerce(obj, attr, value)
if not re.match(self.RE_CERT_ALGO, "%s" % value):
raise ValueError("Cert Algo %s is not a valid Mnemonic or "
"value" % value)
return value
class Any(ovoo_fields.FieldType):
@staticmethod
def coerce(obj, attr, value):

106
designate/objects/rrdata_cert.py Normal file
View File

@ -0,0 +1,106 @@
# Copyright 2021 Cloudification GmbH
#
# Author: cloudification <contact@cloudification.io>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import base64
from designate.exceptions import InvalidObject
from designate.objects.record import Record
from designate.objects.record import RecordList
from designate.objects import base
from designate.objects import fields
@base.DesignateRegistry.register
class CERT(Record):
"""
CERT Resource Record Type
Defined in: RFC4398
"""
fields = {
'cert_type': fields.CertTypeField(),
'key_tag': fields.IntegerFields(minimum=0, maximum=65535),
'cert_algo': fields.CertAlgoField(),
'certificate': fields.StringFields(),
}
def validate_cert_type(self, cert_type):
try:
int_cert_type = int(cert_type)
if int_cert_type < 0 or int_cert_type > 65535:
err = ("Cert type value should be between 0 and 65535")
raise InvalidObject(err)
except ValueError:
# cert type is specified as Mnemonic
VALID_CERTS = ['PKIX', 'SPKI', 'PGP', 'IPKIX', 'ISPKI', 'IPGP',
'ACPKIX', 'IACPKIX', 'URI', 'OID', 'DPKIX', 'DPTR']
if cert_type not in VALID_CERTS:
err = ("Cert type is not valid Mnemonic.")
raise InvalidObject(err)
return cert_type
def validate_cert_algo(self, cert_algo):
try:
int_cert_algo = int(cert_algo)
if int_cert_algo < 0 or int_cert_algo > 255:
err = ("Cert algorithm value should be between 0 and 255")
raise InvalidObject(err)
except ValueError:
# cert algo is specified as Mnemonic
VALID_ALGOS = ['RSAMD5', 'DSA', 'RSASHA1', 'DSA-NSEC3-SHA1',
'RSASHA1-NSEC3-SHA1', 'RSASHA256', 'RSASHA512',
'ECC-GOST', 'ECDSAP256SHA256', 'ECDSAP384SHA384',
'ED25519', 'ED448']
if cert_algo not in VALID_ALGOS:
err = ("Cert algorithm is not valid Mnemonic.")
raise InvalidObject(err)
return cert_algo
def validate_cert_certificate(self, certificate):
try:
chunks = certificate.split(' ')
encoded_chunks = []
for chunk in chunks:
encoded_chunks.append(chunk.encode())
b64 = b''.join(encoded_chunks)
base64.b64decode(b64)
except Exception:
err = ("Cert certificate is not valid.")
raise InvalidObject(err)
return certificate
def _to_string(self):
return ("%(cert_type)s %(key_tag)s %(cert_algo)s "
"%(certificate)s" % self)
def _from_string(self, v):
cert_type, key_tag, cert_algo, certificate = v.split(' ', 3)
self.cert_type = self.validate_cert_type(cert_type)
self.key_tag = int(key_tag)
self.cert_algo = self.validate_cert_algo(cert_algo)
self.certificate = self.validate_cert_certificate(certificate)
RECORD_TYPE = 37
@base.DesignateRegistry.register
class CERTList(RecordList):
LIST_ITEM_TYPE = CERT
fields = {
'objects': fields.ListOfObjectsField('CERT'),
}

29
designate/storage/impl_sqlalchemy/migrate_repo/versions/103_support_cert_records.py Normal file
View File

@ -0,0 +1,29 @@
# Copyright 2021 Cloudification GmbH
#
# Author: cloudification <contact@cloudification.io>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from sqlalchemy import MetaData, Table, Enum
meta = MetaData()
def upgrade(migrate_engine):
meta.bind = migrate_engine
RECORD_TYPES = ['A', 'AAAA', 'CNAME', 'MX', 'SRV', 'TXT', 'SPF', 'NS',
'PTR', 'SSHFP', 'SOA', 'NAPTR', 'CAA', 'CERT']
records_table = Table('recordsets', meta, autoload=True)
records_table.columns.type.alter(name='type', type=Enum(*RECORD_TYPES))

2
designate/storage/impl_sqlalchemy/tables.py
View File

@ -29,7 +29,7 @@ CONF = cfg.CONF
RESOURCE_STATUSES = ['ACTIVE', 'PENDING', 'DELETED', 'ERROR']
RECORD_TYPES = ['A', 'AAAA', 'CNAME', 'MX', 'SRV', 'TXT', 'SPF', 'NS', 'PTR',
'SSHFP', 'SOA', 'NAPTR', 'CAA']
'SSHFP', 'SOA', 'NAPTR', 'CAA', 'CERT']
TASK_STATUSES = ['ACTIVE', 'PENDING', 'DELETED', 'ERROR', 'COMPLETE']
TSIG_ALGORITHMS = ['hmac-md5', 'hmac-sha1', 'hmac-sha224', 'hmac-sha256',

2
designate/tests/test_central/test_service.py
View File

@ -1863,7 +1863,7 @@ class CentralServiceTest(CentralTestCase):
def test_update_recordset_immutable_type(self):
zone = self.create_zone()
# ['A', 'AAAA', 'CNAME', 'MX', 'SRV', 'TXT', 'SPF', 'NS', 'PTR',
# 'SSHFP', 'SOA', 'NAPTR', 'CAA']
# 'SSHFP', 'SOA', 'NAPTR', 'CAA', 'CERT']
# Create a recordset
recordset = self.create_recordset(zone)
cname_recordset = self.create_recordset(zone, type='CNAME')

81
designate/tests/unit/objects/test_cert_object.py Normal file
View File

@ -0,0 +1,81 @@
# Copyright 2021 Cloudification GmbH
#
# Author: cloudification <contact@cloudification.io>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import oslotest.base
from oslo_log import log as logging
from designate import exceptions
from designate import objects
LOG = logging.getLogger(__name__)
class CERTRecordTest(oslotest.base.BaseTestCase):
def test_parse_cert(self):
cert_record = objects.CERT()
cert_record._from_string(
'DPKIX 1 RSASHA256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc=') # noqa
self.assertEqual('DPKIX', cert_record.cert_type)
self.assertEqual(1, cert_record.key_tag)
self.assertEqual('RSASHA256', cert_record.cert_algo)
self.assertEqual('KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc=',
cert_record.certificate)
def test_parse_invalid_cert_type_value(self):
cert_record = objects.CERT()
self.assertRaisesRegex(
exceptions.InvalidObject,
'Cert type value should be between 0 and 65535',
cert_record._from_string,
'99999 1 RSASHA256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc='
)
def test_parse_invalid_cert_type_mnemonic(self):
cert_record = objects.CERT()
self.assertRaisesRegex(
exceptions.InvalidObject,
'Cert type is not valid Mnemonic.',
cert_record._from_string,
'FAKETYPE 1 RSASHA256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc='
)
def test_parse_invalid_cert_algo_value(self):
cert_record = objects.CERT()
self.assertRaisesRegex(
exceptions.InvalidObject,
'Cert algorithm value should be between 0 and 255',
cert_record._from_string,
'DPKIX 1 256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc='
)
def test_parse_invalid_cert_algo_mnemonic(self):
cert_record = objects.CERT()
self.assertRaisesRegex(
exceptions.InvalidObject,
'Cert algorithm is not valid Mnemonic.',
cert_record._from_string,
'DPKIX 1 FAKESHA256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc='
)
def test_parse_invalid_cert_certificate(self):
cert_record = objects.CERT()
self.assertRaisesRegex(
exceptions.InvalidObject,
'Cert certificate is not valid.',
cert_record._from_string,
'DPKIX 1 RSASHA256 KR1L0GbocaIOOim1+qdHtOSrDcOsGiI2NCcxuX2/Tqc'
)

8
doc/source/contributor/sourcedoc/objects.rst
View File

@ -195,3 +195,11 @@ Objects CAA Record
:members:
:undoc-members:
:show-inheritance:
Objects CERT Record
====================
.. automodule:: designate.objects.rrdata_cert
:members:
:undoc-members:
:show-inheritance:

6
releasenotes/notes/CERT_records-eb9b786f480851ff.yaml Normal file
View File

@ -0,0 +1,6 @@
---
features:
- |
CERT recordset type have been added. All users should be able to use this type
from the API and openstack client. This can be disabled (like other record types) by
setting the `[DEFAULT].supported-record-type` config variable in all designate services.