# octavia.amphorae.backends.utils.ip_advertisement

# Source code for octavia.amphorae.backends.utils.ip_advertisement

    # Copyright 2020 Red Hat, Inc. 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.
    import fcntl
    import socket
    import struct

    from oslo_log import log as logging

    from octavia.amphorae.backends.utils import network_namespace
    from octavia.common import constants
    from octavia.common import utils as common_utils

    LOG = logging.getLogger(__name__)

    def garp(interface, ip_address, net_ns=None):
        """Sends a gratuitous ARP for ip_address on the interface.

        :param interface: The interface name to send the GARP on.
        :param ip_address: The IP address to advertise in the GARP.
        :param net_ns: The network namespace to send the GARP from.
        :returns: None
        """
        ARP_ETHERTYPE = 0x0806
        BROADCAST_MAC = b'\xff\xff\xff\xff\xff\xff'

        # Get a socket, optionally inside a network namespace
        garp_socket = None
        if net_ns:
            with network_namespace.NetworkNamespace(net_ns):
                garp_socket = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)
        else:
            garp_socket = socket.socket(socket.AF_PACKET, socket.SOCK_RAW)

        # Bind the socket with the ARP ethertype protocol
        garp_socket.bind((interface, ARP_ETHERTYPE))

        # Get the MAC address of the interface
        source_mac = garp_socket.getsockname()[4]

        garp_msg = [
            struct.pack('!h', 1),                 # Hardware type ethernet
            struct.pack('!h', 0x0800),            # Protocol type IPv4
            struct.pack('!B', 6),                 # Hardware size
            struct.pack('!B', 4),                 # Protocol size
            struct.pack('!h', 1),                 # Opcode request
            source_mac,                    # Sender MAC address
            socket.inet_aton(ip_address),  # Sender IP address
            BROADCAST_MAC,                 # Target MAC address
            socket.inet_aton(ip_address)]  # Target IP address

        garp_ethernet = [
            BROADCAST_MAC,              # Ethernet destination
            source_mac,                 # Ethernet source
            struct.pack('!h', ARP_ETHERTYPE),  # Ethernet type
            b''.join(garp_msg)]         # The GARP message

        garp_socket.send(b''.join(garp_ethernet))
        garp_socket.close()

    def calculate_icmpv6_checksum(packet):
        """Calculate the ICMPv6 checksum for a packet.

        :param packet: The packet bytes to checksum.
        :returns: The checksum integer.
        """
        total = 0

        # Add up 16-bit words
        num_words = len(packet) // 2
        for chunk in struct.unpack("!%sH" % num_words, packet[0:num_words * 2]):
            total += chunk

        # Add any left over byte
        if len(packet) % 2:
            total += packet[-1] << 8

        # Fold 32-bits into 16-bits
        total = (total >> 16) + (total & 0xffff)
        total += total >> 16
        return ~total + 0x10000 & 0xffff

    def neighbor_advertisement(interface, ip_address, net_ns=None):
        """Sends a unsolicited neighbor advertisement for an ip on the interface.

        :param interface: The interface name to send the GARP on.
        :param ip_address: The IP address to advertise in the GARP.
        :param net_ns: The network namespace to send the GARP from.
        :returns: None
        """
        ALL_NODES_ADDR = 'ff02::1'
        SIOCGIFHWADDR = 0x8927

        # Get a socket, optionally inside a network namespace
        na_socket = None
        if net_ns:
            with network_namespace.NetworkNamespace(net_ns):
                na_socket = socket.socket(
                    socket.AF_INET6, socket.SOCK_RAW,
                    socket.getprotobyname(constants.IPV6_ICMP))
        else:
            na_socket = socket.socket(socket.AF_INET6, socket.SOCK_RAW,
                                      socket.getprotobyname(constants.IPV6_ICMP))

        # Per RFC 4861 section 4.4, the hop limit should be 255
        na_socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 255)

        # Bind the socket with the source address
        na_socket.bind((ip_address, 0))

        # Get the byte representation of the MAC address of the interface
        # Note: You can't use getsockname() to get the MAC on this type of socket
        source_mac = fcntl.ioctl(
            na_socket.fileno(), SIOCGIFHWADDR, struct.pack(
                '256s', bytes(interface, 'utf-8')))[18:24]

        # Get the byte representation of the source IP address
        source_ip_bytes = socket.inet_pton(socket.AF_INET6, ip_address)

        icmpv6_na_msg_prefix = [
            struct.pack('!B', 136),   # ICMP Type Neighbor Advertisement
            struct.pack('!B', 0)]     # ICMP Code
        icmpv6_na_msg_postfix = [
            struct.pack('!I', 0xa0000000),  # Flags (Router, Override)
            source_ip_bytes,         # Target address
            struct.pack('!B', 2),    # ICMPv6 option type target link-layer address
            struct.pack('!B', 1),    # ICMPv6 option length
            source_mac]              # ICMPv6 option link-layer address

        # Calculate the ICMPv6 checksum
        icmpv6_pseudo_header = [
            source_ip_bytes,         # Source IP address
            socket.inet_pton(socket.AF_INET6, ALL_NODES_ADDR),  # Destination IP
            struct.pack('!I', 58),   # IPv6 next header (ICMPv6)
            struct.pack('!h', 32)]   # IPv6 payload length
        icmpv6_tmp_chksum = struct.pack('!H', 0)  # Checksum->zeros for calculation
        tmp_chksum_msg = b''.join(icmpv6_pseudo_header + icmpv6_na_msg_prefix +
                                  [icmpv6_tmp_chksum] + icmpv6_pseudo_header)
        checksum = struct.pack('!H', calculate_icmpv6_checksum(tmp_chksum_msg))

        # Build the ICMPv6 unsolicitated neighbor advertisement
        icmpv6_msg = b''.join(icmpv6_na_msg_prefix + [checksum] +
                              icmpv6_na_msg_postfix)

        na_socket.sendto(icmpv6_msg, (ALL_NODES_ADDR, 0, 0, 0))
        na_socket.close()

    def send_ip_advertisement(interface, ip_address, net_ns=None):
        """Send an address advertisement.

        This method will send either GARP (IPv4) or neighbor advertisements (IPv6)
        for the ip address specified.

        :param interface: The interface name to send the advertisement on.
        :param ip_address: The IP address to advertise.
        :param net_ns: The network namespace to send the advertisement from.
        :returns: None
        """
        try:
            if common_utils.is_ipv4(ip_address):
                garp(interface, ip_address, net_ns)
            elif common_utils.is_ipv6(ip_address):
                neighbor_advertisement(interface, ip_address, net_ns)
            else:
                LOG.error('Unknown IP version for address: "%s". Skipping',
                          ip_address)
        except Exception as e:
            LOG.warning('Unable to send address advertisement for address: "%s", '
                        'error: %s. Skipping', ip_address, str(e))

this page last updated: 2026-04-02 20:14:22

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