<commit_before>/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: <NAME>)
 * Copyright 2018 (c) <NAME>, fortiss GmbH
 * Copyright (c) 2020 Fraunhofer IOSB (Author: <NAME>)
 * Copyright (c) 2020 Kalycito Infotech Private Limited
 * Copyright (c) 2021 Linutronix GmbH (Author: <NAME>)
 */

#include <open62541/server_pubsub.h>
#include <open62541/util.h>

#include <open62541/plugin/log_stdout.h>
#include <open62541/plugin/pubsub_udp.h>

#define RECEIVE_MSG_BUFFER_SIZE   4096
static UA_THREAD_LOCAL UA_Byte ReceiveMsgBufferUDP[RECEIVE_MSG_BUFFER_SIZE];


/* UDP multicast network layer specific internal data */
typedef struct {
    int ai_family;                    /* Protocol family for socket. IPv4/IPv6 */
    struct sockaddr_storage ai_addr; /* https://msdn.microsoft.com/de-de/library/windows/desktop/ms740496(v=vs.85).aspx */
    socklen_t ai_addrlen;            /* Address length */
    struct sockaddr_storage intf_addr;
    UA_UInt32 messageTTL;
    UA_Boolean enableLoopback;
    UA_Boolean enableReuse;
    UA_Boolean isMulticast;
} UA_PubSubChannelDataUDPMC;

/**
 * Open communication socket based on the connectionConfig. Protocol specific parameters are
 * provided within the connectionConfig as KeyValuePair.
 * Currently supported options: "ttl" , "loopback", "reuse"
 *
 * @return ref to created channel, NULL on error
 */
static UA_PubSubChannel *
UA_PubSubChannelUDPMC_open(const UA_PubSubConnectionConfig *connectionConfig) {
    UA_initialize_architecture_network();

    if(!UA_Variant_hasScalarType(&connectionConfig->address,
                                &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE])) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Invalid Address.");
        return NULL;
    }

    UA_NetworkAddressUrlDataType *address =
        (UA_NetworkAddressUrlDataType *)connectionConfig->address.data;

    /* Allocate and init memory for the UDP multicast specific internal data */
    UA_PubSubChannelDataUDPMC * channelDataUDPMC = (UA_PubSubChannelDataUDPMC *)
        UA_calloc(1, (sizeof(UA_PubSubChannelDataUDPMC)));
    if(!channelDataUDPMC){
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Out of memory.");
        return NULL;
    }

    /* Set default values */
    UA_PubSubChannelDataUDPMC defaultValues = {0, {0}, 0, {0}, 255, UA_TRUE, UA_TRUE, UA_TRUE};
    memcpy(channelDataUDPMC, &defaultValues, sizeof(UA_PubSubChannelDataUDPMC));
    /* Iterate over the given KeyValuePair parameters */
    UA_String ttlParam = UA_STRING("ttl");
    UA_String loopbackParam = UA_STRING("loopback");
    UA_String reuseParam = UA_STRING("reuse");
#ifdef __linux__
    UA_String socketPriorityParam = UA_STRING("sockpriority");
    UA_UInt32  *socketPriority = NULL;
#endif
    for(size_t i = 0; i < connectionConfig->connectionPropertiesSize; i++) {
        UA_KeyValuePair *prop = &connectionConfig->connectionProperties[i];
        if(UA_String_equal(&prop->key.name, &ttlParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_UINT32])){
                channelDataUDPMC->messageTTL = *(UA_UInt32*)prop->value.data;
            }
        } else if(UA_String_equal(&prop->key.name, &loopbackParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {
                channelDataUDPMC->enableLoopback = *(UA_Boolean*)prop->value.data;
            }
        } else if(UA_String_equal(&prop->key.name, &reuseParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {
                channelDataUDPMC->enableReuse = *(UA_Boolean*)prop->value.data;
            }
#ifdef __linux__
        } else if(UA_String_equal(&prop->key.name, &socketPriorityParam)){
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_UINT32])){
                socketPriority = (UA_UInt32 *) UA_malloc(sizeof(UA_UInt32));
                UA_UInt32_copy((UA_UInt32 *) prop->value.data, socketPriority);
            }
#endif
        } else {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                           "PubSub Connection creation. Unknown connection parameter.");
        }
    }

    UA_PubSubChannel *newChannel = (UA_PubSubChannel *) UA_calloc(1, sizeof(UA_PubSubChannel));
    if(!newChannel) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Out of memory.");
        UA_free(channelDataUDPMC);
        return NULL;
    }
    struct addrinfo hints, *rp, *requestResult = NULL;
    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_DGRAM;
    hints.ai_flags = 0;
    hints.ai_protocol = 0;

    UA_String hostname = UA_STRING_NULL;
    UA_String path = UA_STRING_NULL;
    UA_UInt16 networkPort = 0;
    UA_StatusCode res = UA_parseEndpointUrl(&address->url, &hostname, &networkPort, &path);
    if(res != UA_STATUSCODE_GOOD){
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Invalid URL.");
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }
    if(hostname.length > 512) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. URL maximum length is 512.");
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    UA_STACKARRAY(char, addressAsChar, sizeof(char) * hostname.length +1);
    memcpy(addressAsChar, hostname.data, hostname.length);
    addressAsChar[hostname.length] = 0;
    char port[6];
    sprintf(port, "%u", networkPort);

    int error = UA_getaddrinfo(addressAsChar, port, &hints, &requestResult);
    if(error) {
        errno = error;
        UA_LOG_SOCKET_ERRNO_GAI_WRAP(
            UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_NETWORK,
                         "PubSub Connection creation failed. Internal error: "
                         "getaddrinfo lookup of %s failed with error %s",
                         addressAsChar, errno_str));
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }


    /* Check if the ip address is a multicast address */
    if(requestResult->ai_family == PF_INET) {
        struct in_addr imr_interface;
        memset(&imr_interface, 0, sizeof(imr_interface));
        if(UA_inet_pton(AF_INET, addressAsChar, &imr_interface) == 1 &&
           (UA_ntohl(imr_interface.s_addr) & 0xF0000000) != 0xE0000000) {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                "PubSub Connection is created for a unicast address (IPv4)");
            channelDataUDPMC->isMulticast = UA_FALSE;
        }
    } else {
        struct in6_addr imr_interface;
        memset(&imr_interface, 0, sizeof(imr_interface));
        if(UA_inet_pton(AF_INET6, addressAsChar, &imr_interface) == 1 &&
           (imr_interface.s6_addr[0] != 0xFF)) {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                "PubSub Connection is created for a unicast address (IPv6)");
            channelDataUDPMC->isMulticast = UA_FALSE;
        }
    }

    for(rp = requestResult; rp != NULL; rp = rp->ai_next){
        newChannel->sockfd = UA_socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if(newChannel->sockfd != UA_INVALID_SOCKET)
            break; /*success*/
    }
    if(!rp) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Internal error.");
        UA_freeaddrinfo(requestResult);
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    channelDataUDPMC->ai_family = rp->ai_family;
    channelDataUDPMC->ai_addrlen = rp->ai_addrlen;<commit_msg>refactor(plugin): Cosmetic improvements in ua_pubsub_udp.c<commit_after>/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: <NAME>)
 * Copyright 2018 (c) <NAME>, fortiss GmbH
 * Copyright (c) 2020 Fraunhofer IOSB (Author: <NAME>)
 * Copyright (c) 2020 Kalycito Infotech Private Limited
 * Copyright (c) 2021 Linutronix GmbH (Author: <NAME>)
 */

#include <open62541/server_pubsub.h>
#include <open62541/util.h>

#include <open62541/plugin/log_stdout.h>
#include <open62541/plugin/pubsub_udp.h>

#define RECEIVE_MSG_BUFFER_SIZE   4096
static UA_THREAD_LOCAL UA_Byte ReceiveMsgBufferUDP[RECEIVE_MSG_BUFFER_SIZE];


/* UDP multicast network layer specific internal data */
typedef struct {
    int ai_family;                   /* Protocol family for socket. IPv4/IPv6 */
    struct sockaddr_storage ai_addr; /* https://msdn.microsoft.com/de-de/library/windows/desktop/ms740496(v=vs.85).aspx */
    socklen_t ai_addrlen;            /* Address length */
    struct sockaddr_storage intf_addr;
    UA_UInt32 messageTTL;
    UA_Boolean enableLoopback;
    UA_Boolean enableReuse;
    UA_Boolean isMulticast;
} UA_PubSubChannelDataUDPMC;

/**
 * Open communication socket based on the connectionConfig. Protocol specific parameters are
 * provided within the connectionConfig as KeyValuePair.
 * Currently supported options: "ttl" , "loopback", "reuse"
 *
 * @return ref to created channel, NULL on error
 */
static UA_PubSubChannel *
UA_PubSubChannelUDPMC_open(const UA_PubSubConnectionConfig *connectionConfig) {
    UA_initialize_architecture_network();

    if(!UA_Variant_hasScalarType(&connectionConfig->address,
                                &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE])) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Invalid Address.");
        return NULL;
    }

    UA_NetworkAddressUrlDataType *address =
        (UA_NetworkAddressUrlDataType *)connectionConfig->address.data;

    /* Allocate and init memory for the UDP multicast specific internal data */
    UA_PubSubChannelDataUDPMC * channelDataUDPMC = (UA_PubSubChannelDataUDPMC *)
        UA_calloc(1, (sizeof(UA_PubSubChannelDataUDPMC)));
    if(!channelDataUDPMC){
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Out of memory.");
        return NULL;
    }

    /* Set default values */
    UA_PubSubChannelDataUDPMC defaultValues =
        {0, {0}, 0, {0}, 255, true, true, true};
    memcpy(channelDataUDPMC, &defaultValues, sizeof(UA_PubSubChannelDataUDPMC));
    /* Iterate over the given KeyValuePair parameters */
    UA_String ttlParam = UA_STRING("ttl");
    UA_String loopbackParam = UA_STRING("loopback");
    UA_String reuseParam = UA_STRING("reuse");
#ifdef __linux__
    UA_String socketPriorityParam = UA_STRING("sockpriority");
    UA_UInt32  *socketPriority = NULL;
#endif
    for(size_t i = 0; i < connectionConfig->connectionPropertiesSize; i++) {
        UA_KeyValuePair *prop = &connectionConfig->connectionProperties[i];
        if(UA_String_equal(&prop->key.name, &ttlParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_UINT32])){
                channelDataUDPMC->messageTTL = *(UA_UInt32*)prop->value.data;
            }
        } else if(UA_String_equal(&prop->key.name, &loopbackParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {
                channelDataUDPMC->enableLoopback = *(UA_Boolean*)prop->value.data;
            }
        } else if(UA_String_equal(&prop->key.name, &reuseParam)) {
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {
                channelDataUDPMC->enableReuse = *(UA_Boolean*)prop->value.data;
            }
#ifdef __linux__
        } else if(UA_String_equal(&prop->key.name, &socketPriorityParam)){
            if(UA_Variant_hasScalarType(&prop->value, &UA_TYPES[UA_TYPES_UINT32])){
                socketPriority = (UA_UInt32 *) UA_malloc(sizeof(UA_UInt32));
                UA_UInt32_copy((UA_UInt32 *) prop->value.data, socketPriority);
            }
#endif
        } else {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                           "PubSub Connection creation. Unknown connection parameter.");
        }
    }

    UA_PubSubChannel *newChannel = (UA_PubSubChannel *)
        UA_calloc(1, sizeof(UA_PubSubChannel));
    if(!newChannel) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Out of memory.");
        UA_free(channelDataUDPMC);
        return NULL;
    }
    struct addrinfo hints, *rp, *requestResult = NULL;
    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_DGRAM;
    hints.ai_flags = 0;
    hints.ai_protocol = 0;

    UA_String hostname = UA_STRING_NULL;
    UA_String path = UA_STRING_NULL;
    UA_UInt16 networkPort = 0;
    UA_StatusCode res = UA_parseEndpointUrl(&address->url, &hostname, &networkPort, &path);
    if(res != UA_STATUSCODE_GOOD){
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Invalid URL.");
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    if(hostname.length >= 512) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. URL maximum length is 512.");
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    char addressAsChar[512];
    memcpy(addressAsChar, hostname.data, hostname.length);
    addressAsChar[hostname.length] = 0;

    char port[6];
    sprintf(port, "%u", networkPort);

    int error = UA_getaddrinfo(addressAsChar, port, &hints, &requestResult);
    if(error) {
        errno = error;
        UA_LOG_SOCKET_ERRNO_GAI_WRAP(
            UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_NETWORK,
                         "PubSub Connection creation failed. Internal error: "
                         "getaddrinfo lookup of %s failed with error %s",
                         addressAsChar, errno_str));
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    /* Check if the ip address is a multicast address */
    if(requestResult->ai_family == PF_INET) {
        struct in_addr imr_interface;
        memset(&imr_interface, 0, sizeof(imr_interface));
        if(UA_inet_pton(AF_INET, addressAsChar, &imr_interface) == 1 &&
           (UA_ntohl(imr_interface.s_addr) & 0xF0000000) != 0xE0000000) {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                "PubSub Connection is created for a unicast address (IPv4)");
            channelDataUDPMC->isMulticast = false;
        }
    } else {
        struct in6_addr imr_interface;
        memset(&imr_interface, 0, sizeof(imr_interface));
        if(UA_inet_pton(AF_INET6, addressAsChar, &imr_interface) == 1 &&
           (imr_interface.s6_addr[0] != 0xFF)) {
            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                "PubSub Connection is created for a unicast address (IPv6)");
            channelDataUDPMC->isMulticast = false;
        }
    }

    for(rp = requestResult; rp != NULL; rp = rp->ai_next){
        newChannel->sockfd = UA_socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if(newChannel->sockfd != UA_INVALID_SOCKET)
            break; /*success*/
    }
    if(!rp) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
                     "PubSub Connection creation failed. Internal error.");
        UA_freeaddrinfo(requestResult);
        UA_free(channelDataUDPMC);
        UA_free(newChannel);
        return NULL;
    }

    channelDataUDPMC->ai_family = rp->ai_family;
    channelDataUDPMC->ai_addrlen = rp->ai_addrlen;
{'fraction_non_alphanumeric': 0.10203704843374976, 'fraction_numerical': 0.016370238168502676, 'mean_word_length': 3.0192926045016075, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 87, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1424712', 'n_tokens_mistral': 5549, 'n_tokens_neox': 5207, 'n_words': 1034}