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