Add dns constraints

This adds constraints to be used with properties of neutron
resources for internal/external dns resolution.

Change-Id: I728eec876b9f5e12b92ee8283c0d1a7610d7ed76
Blueprint: neutron-dns-resolution
This commit is contained in:
rabi 2016-07-07 17:08:40 +05:30 committed by Rabi Mishra
parent 97a7e96e24
commit 5797d34ccd
4 changed files with 203 additions and 0 deletions

View File

@ -12,6 +12,13 @@
# under the License.
import netaddr
import re
from heat.common.i18n import _
DNS_LABEL_MAX_LEN = 63
DNS_LABEL_REGEX = "[a-z0-9-]{1,%d}$" % DNS_LABEL_MAX_LEN
FQDN_MAX_LEN = 255
def is_prefix_subset(orig_prefixes, new_prefixes):
@ -24,3 +31,32 @@ def is_prefix_subset(orig_prefixes, new_prefixes):
orig_set = netaddr.IPSet(orig_prefixes)
new_set = netaddr.IPSet(new_prefixes)
return orig_set.issubset(new_set)
def validate_dns_format(data):
if not data:
return
trimmed = data if not data.endswith('.') else data[:-1]
if len(trimmed) > FQDN_MAX_LEN:
raise ValueError(
_("'%(data)s' exceeds the %(max_len)s character FQDN limit") % {
'data': trimmed,
'max_len': FQDN_MAX_LEN})
names = trimmed.split('.')
for name in names:
if not name:
raise ValueError(_("Encountered an empty component."))
if name.endswith('-') or name.startswith('-'):
raise ValueError(
_("Name '%s' must not start or end with a hyphen.") % name)
if not re.match(DNS_LABEL_REGEX, name):
raise ValueError(
_("Name '%(name)s' must be 1-%(max_len)s characters long, "
"each of which can only be alphanumeric or "
"a hyphen.") % {'name': name,
'max_len': DNS_LABEL_MAX_LEN})
# RFC 1123 hints that a Top Level Domain(TLD) can't be all numeric.
# Last part is a TLD, if it's a FQDN.
if (data.endswith('.') and len(names) > 1
and re.match("^[0-9]+$", names[-1])):
raise ValueError(_("TLD '%s' must not be all numeric.") % names[-1])

View File

@ -21,6 +21,7 @@ from oslo_utils import netutils
from oslo_utils import timeutils
from heat.common.i18n import _
from heat.common import netutils as heat_netutils
from heat.engine import constraints
@ -44,6 +45,59 @@ class MACConstraint(constraints.BaseCustomConstraint):
return netaddr.valid_mac(value)
class DNSNameConstraint(constraints.BaseCustomConstraint):
def validate(self, value, context):
try:
heat_netutils.validate_dns_format(value)
except ValueError as ex:
self._error_message = ("'%(value)s' not in valid format."
" Reason: %(reason)s") % {
'value': value,
'reason': six.text_type(ex)}
return False
return True
class RelativeDNSNameConstraint(DNSNameConstraint):
def validate(self, value, context):
if not value:
return True
if value.endswith('.'):
self._error_message = _("'%s' is a FQDN. It should be a "
"relative domain name.") % value
return False
length = len(value)
if length > heat_netutils.FQDN_MAX_LEN - 3:
self._error_message = _("'%(value)s' contains '%(length)s' "
"characters. Adding a domain name will "
"cause it to exceed the maximum length "
"of a FQDN of '%(max_len)s'.") % {
"value": value,
"length": length,
"max_len": heat_netutils.FQDN_MAX_LEN}
return False
return super(RelativeDNSNameConstraint, self).validate(value, context)
class DNSDomainConstraint(DNSNameConstraint):
def validate(self, value, context):
if not value:
return True
if not super(DNSDomainConstraint, self).validate(value, context):
return False
if not value.endswith('.'):
self._error_message = ("'%s' must end with '.'.") % value
return False
return True
class CIDRConstraint(constraints.BaseCustomConstraint):
def _validate_whitespace(self, data):

View File

@ -195,3 +195,113 @@ class TimezoneConstraintTest(common.HeatTestCase):
def test_validation_none(self):
self.assertTrue(self.constraint.validate(None, self.ctx))
class DNSNameConstraintTest(common.HeatTestCase):
def setUp(self):
super(DNSNameConstraintTest, self).setUp()
self.ctx = utils.dummy_context()
self.constraint = cc.DNSNameConstraint()
def test_validation(self):
self.assertTrue(self.constraint.validate("openstack.org.", self.ctx))
def test_validation_error_hyphen(self):
dns_name = "-openstack.org"
expected = ("'%s' not in valid format. Reason: Name "
"'%s' must not start or end with a "
"hyphen.") % (dns_name, dns_name.split('.')[0])
self.assertFalse(self.constraint.validate(dns_name, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_error_empty_component(self):
dns_name = ".openstack.org"
expected = ("'%s' not in valid format. Reason: "
"Encountered an empty component.") % dns_name
self.assertFalse(self.constraint.validate(dns_name, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_error_special_char(self):
dns_name = "$openstack.org"
expected = ("'%s' not in valid format. Reason: Name "
"'%s' must be 1-63 characters long, each "
"of which can only be alphanumeric or a "
"hyphen.") % (dns_name, dns_name.split('.')[0])
self.assertFalse(self.constraint.validate(dns_name, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_error_tld_allnumeric(self):
dns_name = "openstack.123."
expected = ("'%s' not in valid format. Reason: TLD "
"'%s' must not be all numeric.") % (dns_name,
dns_name.split('.')[1])
self.assertFalse(self.constraint.validate(dns_name, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_none(self):
self.assertTrue(self.constraint.validate(None, self.ctx))
class DNSDomainConstraintTest(common.HeatTestCase):
def setUp(self):
super(DNSDomainConstraintTest, self).setUp()
self.ctx = utils.dummy_context()
self.constraint = cc.DNSDomainConstraint()
def test_validation(self):
self.assertTrue(self.constraint.validate("openstack.org.", self.ctx))
def test_validation_error_no_end_period(self):
dns_domain = "openstack.org"
expected = ("'%s' must end with '.'.") % dns_domain
self.assertFalse(self.constraint.validate(dns_domain, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_none(self):
self.assertTrue(self.constraint.validate(None, self.ctx))
class FIPDNSNameConstraintTest(common.HeatTestCase):
def setUp(self):
super(FIPDNSNameConstraintTest, self).setUp()
self.ctx = utils.dummy_context()
self.constraint = cc.RelativeDNSNameConstraint()
def test_validation(self):
self.assertTrue(self.constraint.validate("myvm.openstack", self.ctx))
def test_validation_error_end_period(self):
dns_name = "myvm.openstack."
expected = ("'%s' is a FQDN. It should be a relative "
"domain name.") % dns_name
self.assertFalse(self.constraint.validate(dns_name, self.ctx))
self.assertEqual(
expected,
six.text_type(self.constraint._error_message)
)
def test_validation_none(self):
self.assertTrue(self.constraint.validate(None, self.ctx))

View File

@ -80,6 +80,9 @@ heat.clients =
heat.constraints =
# common constraints
cron_expression = heat.engine.constraint.common_constraints:CRONExpressionConstraint
dns_domain = heat.engine.constraint.common_constraints:DNSDomainConstraint
dns_name = heat.engine.constraint.common_constraints:DNSNameConstraint
rel_dns_name = heat.engine.constraint.common_constraints:RelativeDNSNameConstraint
ip_addr = heat.engine.constraint.common_constraints:IPConstraint
iso_8601 = heat.engine.constraint.common_constraints:ISO8601Constraint
mac_addr = heat.engine.constraint.common_constraints:MACConstraint