# designate.backend.impl_nsd4

# Source code for designate.backend.impl_nsd4

    # Copyright (C) 2013 eNovance SAS <licensing@enovance.com>
    # Copyright 2014 eBay Inc.
    # Copyright 2015 Zetta.IO.
    #
    # Author: Ron Rickard <rrickard@ebay.com>
    # Author: Artom Lifshitz <artom.lifshitz@enovance.com>
    # Author: Dag Stenstad <dag@stenstad.net>
    #
    # 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 random
    import socket
    import ssl

    from oslo_log import log as logging

    from designate.backend import base
    from designate.common import constants
    from designate.common import crypto_utils
    import designate.conf
    from designate import exceptions

    CONF = designate.conf.CONF

    LOG = logging.getLogger(__name__)

    class NSD4Backend(base.Backend):

        __backend_status__ = 'untested'

        __plugin_name__ = 'nsd4'
        NSDCT_VERSION = 'NSDCT1'

        def __init__(self, target):
            super().__init__(target)

            self.host = self.options.get('host', '127.0.0.1')
            self.port = int(self.options.get('port', 8952))
            self.certfile = self.options.get('certfile',
                                             '/etc/nsd/nsd_control.pem')
            self.keyfile = self.options.get('keyfile',
                                            '/etc/nsd/nsd_control.key')
            self.pattern = self.options.get('pattern', 'slave')

            check_mode = CONF['pqc'].check_mode
            if check_mode != constants.PQC_MODE_DISABLED:
                cert_paths = [
                    p for p in [self.certfile, self.options.get('ca_certs')]
                    if p
                ]
                crypto_utils.check_pqc_compliance(
                    cert_paths=cert_paths,
                    check_mode=check_mode,
                    component_name=self.__plugin_name__
                )

        def _create_ssl_context(self):
            """Create and configure SSL context with backend settings."""
            context = ssl.create_default_context()
            context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)

            tls_min = CONF['pqc'].tls_minimum_version
            if tls_min:
                crypto_utils.check_tls_version_support(tls_min)
                context.minimum_version = crypto_utils.TLS_VERSION_MAP[tls_min]

            verify_ssl = (
                self.options.get('verify_ssl', 'true').lower() == 'true'
            )
            check_hostname = (
                self.options.get('check_hostname', 'true').lower() == 'true'
            )
            ca_certs = self.options.get('ca_certs', None)
            check_mode = CONF['pqc'].check_mode

            context.check_hostname = check_hostname
            if not check_hostname and check_mode != constants.PQC_MODE_DISABLED:
                LOG.warning(
                    'NSD4 backend: TLS hostname verification is DISABLED '
                    '(check_hostname=false). This allows man-in-the-middle '
                    'attacks even when certificate verification is enabled.'
                )
                if check_mode == constants.PQC_MODE_STRICT:
                    raise exceptions.ConfigurationError(
                        'NSD4 backend: Hostname verification cannot be '
                        'disabled when pqc check_mode is "strict".'
                    )

            if verify_ssl:
                context.verify_mode = ssl.CERT_REQUIRED
                if ca_certs:
                    context.load_verify_locations(cafile=ca_certs)
            else:
                if check_mode != constants.PQC_MODE_DISABLED:
                    LOG.warning(
                        'NSD4 backend: TLS certificate verification is '
                        'DISABLED (verify_ssl=false). This allows '
                        'man-in-the-middle attacks and negates any PQC '
                        'key exchange protection.'
                    )
                    if check_mode == constants.PQC_MODE_STRICT:
                        raise exceptions.ConfigurationError(
                            'NSD4 backend: Certificate verification '
                            'cannot be disabled when pqc check_mode is '
                            '"strict".'
                        )
                context.verify_mode = ssl.CERT_NONE

            return context

        def _command(self, command):
            """Execute a command on the NSD4 control channel."""
            context = self._create_ssl_context()
            sock_addr = (self.host, self.port)

            with (
                socket.create_connection(sock_addr) as raw_sock,
                context.wrap_socket(raw_sock, server_hostname=self.host) as sock,
                sock.makefile(mode='rw') as stream
            ):
                stream.write(f'{self.NSDCT_VERSION} {command}\n')
                stream.flush()
                return stream.read()

        def _execute_nsd4(self, command):
            try:
                LOG.debug('Executing NSD4 control call: %s on %s',
                          command, self.host)
                result = self._command(command)
            except (ssl.SSLError, OSError) as e:
                LOG.debug('NSD4 control call failure: %s' % e)
                raise exceptions.Backend(e)
            if result.rstrip("\n") != 'ok':
                raise exceptions.Backend(result)

        def create_zone(self, context, zone):
            LOG.debug('Create Zone')
            masters = []
            for master in self.masters:
                host = master['host']
                port = master['port']
                masters.append(f'{host} port {port}')

            # Ensure different MiniDNS instances are targeted for AXFRs
            random.shuffle(masters)

            command = 'addzone {} {}'.format(zone['name'], self.pattern)

            try:
                self._execute_nsd4(command)
            except exceptions.Backend as e:
                # If create fails because the zone exists, don't reraise
                if "already exists" not in str(e):
                    raise

        def delete_zone(self, context, zone, zone_params=None):
            LOG.debug('Delete Zone')
            command = 'delzone %s' % zone['name']

            try:
                self._execute_nsd4(command)
            except exceptions.Backend as e:
                # If zone is already deleted, don't reraise
                if "not found" not in str(e):
                    raise

this page last updated: 2026-08-19 02:25:39

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug