# octavia.common.utils

# Source code for octavia.common.utils

    # Copyright 2011, VMware, Inc., 2014 A10 Networks
    # All Rights Reserved.
    #
    #    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.
    #
    # Borrowed from nova code base, more utilities will be added/borrowed as and
    # when needed.

    """Utilities and helper functions."""

    import base64
    import hashlib
    import ipaddress
    import re
    import socket
    import typing

    from cryptography import fernet
    from oslo_config import cfg
    from oslo_log import log as logging
    from oslo_utils import excutils
    from stevedore import driver as stevedore_driver

    from octavia.common import constants

    if typing.TYPE_CHECKING:
        from octavia.network import base as network_base

    CONF = cfg.CONF

    LOG = logging.getLogger(__name__)

    def get_hostname():
        return socket.gethostname()

    def base64_sha1_string(string_to_hash):
        """Get a b64-encoded sha1 hash of a string. Not intended to be secure!"""
        # TODO(rm_work): applying nosec here because this is not intended to be
        # secure, it's just a way to get a consistent ID. Changing this would
        # break backwards compatibility with existing loadbalancers.
        hash_str = hashlib.sha1(string_to_hash.encode('utf-8')).digest()  # nosec
        b64_str = base64.b64encode(hash_str, str.encode('_-', 'ascii'))
        b64_sha1 = b64_str.decode('UTF-8')
        # https://github.com/haproxy/haproxy/issues/644
        return re.sub(r"^-", "x", b64_sha1)

    def get_amphora_driver():
        amphora_driver = stevedore_driver.DriverManager(
            namespace='octavia.amphora.drivers',
            name=CONF.controller_worker.amphora_driver,
            invoke_on_load=True
        ).driver
        return amphora_driver

    def get_network_driver() -> 'network_base.AbstractNetworkDriver':
        CONF.import_group('controller_worker', 'octavia.common.config')
        network_driver = stevedore_driver.DriverManager(
            namespace='octavia.network.drivers',
            name=CONF.controller_worker.network_driver,
            invoke_on_load=True
        ).driver
        return network_driver

    def ip_version(ip_address):
        ip = ipaddress.ip_address(ip_address)
        return ip.version

    def is_ipv4(ip_address):
        """Check if ip address is IPv4 address."""
        ip = ipaddress.ip_address(ip_address)
        return ip.version == 4

    def is_ipv6(ip_address):
        """Check if ip address is IPv6 address."""
        ip = ipaddress.ip_address(ip_address)
        return ip.version == 6

    def is_cidr_ipv6(cidr):
        """Check if CIDR is IPv6 address with subnet prefix."""
        ip = ipaddress.ip_network(cidr, strict=False)
        return ip.version == 6

    def is_ipv6_lla(ip_address):
        """Check if ip address is IPv6 link local address."""
        ip = ipaddress.ip_address(ip_address)
        return ip.version == 6 and ip.is_link_local

    def ip_port_str(ip_address, port):
        """Return IP port as string representation depending on address family."""
        ip = ipaddress.ip_address(ip_address)
        if ip.version == 4:
            return f"{ip}:{port}"
        return f"[{ip}]:{port}"

    def netmask_to_prefix(netmask):
        return ipaddress.ip_network(f'0.0.0.0/{netmask}', strict=False).prefixlen

    def ip_netmask_to_cidr(ip, netmask):
        return ipaddress.ip_network(f'{ip}/{netmask}', strict=False).with_prefixlen

    def get_vip_security_group_name(loadbalancer_id):
        if loadbalancer_id:
            return constants.VIP_SECURITY_GROUP_PREFIX + loadbalancer_id
        return None

    def get_compatible_value(value):
        if isinstance(value, str):
            value = value.encode('utf-8')
        return value

    def _get_compatible_server_certs_key_passphrases():
        key_opts = CONF.certificates.server_certs_key_passphrase
        keys = []
        for key_opt in key_opts:
            key = str(key_opt)
            if isinstance(key, str):
                key = key.encode('utf-8')
            keys.append(
                base64.urlsafe_b64encode(key))
        return keys

    def get_server_certs_key_passphrases_fernet() -> fernet.MultiFernet:
        """Get a cryptography.MultiFernet with loaded keys."""
        keys = [
            fernet.Fernet(x) for x in
            _get_compatible_server_certs_key_passphrases()]
        return fernet.MultiFernet(keys)

    def subnet_ip_availability(nw_ip_avail, subnet_id, req_num_ips):
        for subnet in nw_ip_avail.subnet_ip_availability:
            if subnet['subnet_id'] == subnet_id:
                return subnet['total_ips'] - subnet['used_ips'] >= req_num_ips
        return None

    def b(s):
        return s.encode('utf-8')

    def expand_expected_codes(codes):
        """Expand the expected code string in set of codes.

        200-204 -> 200, 201, 202, 204
        200, 203 -> 200, 203
        """
        retval = set()
        codes = re.split(', *', codes)
        for code in codes:
            if not code:
                continue
            if '-' in code:
                low, hi = code.split('-')[:2]
                retval.update(
                    str(i) for i in range(int(low), int(hi) + 1))
            else:
                retval.add(code)
        return retval

    class exception_logger:
        """Wrap a function and log raised exception

        :param logger: the logger to log the exception default is LOG.exception

        :returns: origin value if no exception raised; re-raise the exception if
                  any occurred

        """

        def __init__(self, logger=None):
            self.logger = logger

        def __call__(self, func):
            if self.logger is None:
                _LOG = logging.getLogger(func.__module__)
                self.logger = _LOG.exception

            def call(*args, **kwargs):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    with excutils.save_and_reraise_exception():
                        self.logger(e)
                return None
            return call

    def map_protocol_to_nftable_protocol(rule_dict):
        rule_dict[constants.PROTOCOL] = (
            constants.L4_PROTOCOL_MAP[rule_dict[constants.PROTOCOL]])
        return rule_dict

this page last updated: 2026-04-13 15:26: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