client_software.cpp

#include <open62541/client.h>
#include <open62541/client_config_default.h>
#include <open62541/client_highlevel.h>
#include <open62541/client_subscriptions.h>
#include <open62541/plugin/log_stdout.h>
#include <signal.h>
#include <stdlib.h>
#include <string.h>

#include <array>
#include <atomic>
#include <chrono>
#include <functional>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <span>
#include <string>
#include <string_view>
#include <thread>
#include <variant>
#include <vector>

#include "cert_parser.h"
extern "C" {
#include "fips202.h"
}
#include "mldsa.h"

std::atomic<long> measure_counter{1};
using namespace std::literals;

using ReceivingDataType = std::variant<std::vector<uint8_t>, std::string>;

static volatile UA_Boolean running = true;

std::string get_current_time_us() {
    auto now = std::chrono::system_clock::now();
    auto time_t = std::chrono::system_clock::to_time_t(now);
    auto us = std::chrono::duration_cast<std::chrono::microseconds>(
                  now.time_since_epoch()) %
              1000000;

    std::ostringstream oss;
    oss << "[" << std::put_time(std::localtime(&time_t), "%H:%M:%S");
    oss << "." << std::setfill('0') << std::setw(6) << us.count() << "]";

    return oss.str();
}

std::string signing_time;

//=============================
// MEASUREMENT PURPOSE
//=============================
std::chrono::_V2::system_clock::time_point signing_start_time;
int64_t signing_timestamp;
//=======================

/**
 * @brief Signal handling
 *
 * @param sign
 */
static void stopHandler(int sign) {
    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                "Receiving software interrupt. Exiting...");
    running = false;
}

// Structure for holding sensor data with thread safety
class SensorDataContainer {
   private:
    mutable std::mutex mtx_;
    std::string rawDataString_;
    std::vector<uint8_t> signature_;
    std::atomic<bool> rawDataUpdated_{false};
    std::atomic<bool> signatureUpdated_{false};
    std::atomic<bool> dataReady_{false};

   public:
    void updateRawData(const std::string &data) {
        std::lock_guard<std::mutex> lock(mtx_);
        rawDataString_ = data;
        rawDataUpdated_ = true;
        checkDataReady();
    }

    void updateSignature(const std::vector<uint8_t> &sig) {
        std::lock_guard<std::mutex> lock(mtx_);
        signature_ = sig;
        signatureUpdated_ = true;
        checkDataReady();
    }

    bool isDataReady() const { return dataReady_.load(); }

    std::pair<std::string, std::vector<uint8_t>> getData() {
        std::lock_guard<std::mutex> lock(mtx_);
        auto result = std::make_pair(rawDataString_, signature_);
        // Reset flags after consuming data
        rawDataUpdated_ = false;
        signatureUpdated_ = false;
        dataReady_ = false;
        return result;
    }

   private:
    void checkDataReady() {
        if (rawDataUpdated_.load() && signatureUpdated_.load()) {
            dataReady_ = true;
        }
    }
};

// Global sensor data container
static SensorDataContainer g_sensorData;

// Global signature handler and crypto objects
static CertParser *g_cert = nullptr;
static MLDSA<ParamSet_65> *g_sigHndl = nullptr;
static std::array<uint8_t, ParamSet_65::CRYPTO_PUBLICKEYBYTES> g_publicKey;

/**
 * @brief Callback for raw data changes
 */
static void rawDataChangeCallback(UA_Client *client, UA_UInt32 subId,
                                  void *subContext, UA_UInt32 monId,
                                  void *monContext, UA_DataValue *value) {
    if (value->hasValue && value->value.type == &UA_TYPES[UA_TYPES_STRING]) {
        UA_String *transmitString = (UA_String *)value->value.data;
        std::string rawData(reinterpret_cast<char *>(transmitString->data),
                            transmitString->length);

        // ==================== raw data string mesurement  ==============

        // signing_start_time = std::chrono::high_resolution_clock::now();
        // signing_timestamp =
        //     std::chrono::duration_cast<std::chrono::microseconds>(
        //         signing_start_time.time_since_epoch())
        //         .count();
        // signing_time = get_current_time_us();
        // signing_time += " ";
        // signing_time += std::to_string(signing_timestamp);
        // ==========================================================
        g_sensorData.updateRawData(rawData);

        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                    "Raw data updated: %.*s", (int)transmitString->length,
                    transmitString->data);
    } else {
        UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                       "Invalid raw data received");
    }
}

/**
 * @brief Callback for signature changes
 */
static void signatureChangeCallback(UA_Client *client, UA_UInt32 subId,
                                    void *subContext, UA_UInt32 monId,
                                    void *monContext, UA_DataValue *value) {
    if (value->hasValue && value->value.type == &UA_TYPES[UA_TYPES_BYTE] &&
        !UA_Variant_isScalar(&value->value)) {
        UA_Byte *dataPtr = (UA_Byte *)value->value.data;
        // modify this line for changing the signature size
        // size_t sigLength = value->value.arrayLength;
        size_t sigLength = ParamSet_65::CRYPTO_BYTES;

        // // ==================== signature mesurement  ==============

        signing_start_time = std::chrono::high_resolution_clock::now();
        signing_timestamp =
            std::chrono::duration_cast<std::chrono::microseconds>(
                signing_start_time.time_since_epoch())
                .count();
        signing_time = get_current_time_us();
        signing_time += " ";
        signing_time += std::to_string(signing_timestamp);
        // // ==========================================================

        std::vector<uint8_t> signature(dataPtr, dataPtr + sigLength);
        g_sensorData.updateSignature(signature);

        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                    "Signature updated (length: %lu)", sigLength);
    } else {
        UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                       "Invalid signature data received");
    }
}

/**
 * @brief Process and verify sensor data when both raw data and signature are
 * available
 */
void processSensorData() {
    auto [rawDataString, signature] = g_sensorData.getData();

    if (rawDataString.empty() || signature.empty()) {
        UA_LOG_WARNING(
            UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
            "Incomplete data received (raw: %lu bytes, sig: %lu bytes)",
            rawDataString.size(), signature.size());
        return;
    }

    // Perform signature verification
    std::array<uint8_t, 128> msgHashing;
    msgHashing.fill(0);

    std::span<uint8_t> dataIn(
        reinterpret_cast<uint8_t *>(const_cast<char *>(rawDataString.data())),
        rawDataString.size());

    keccak_state state;
    shake128_init(&state);
    shake128_absorb(&state, dataIn.data(), dataIn.size());
    shake128_finalize(&state);
    shake128_squeeze(msgHashing.data(), 128, &state);

    int sigVerifyResult =
        g_sigHndl->Verify(signature.data(), signature.size(),    // signature
                          msgHashing.data(), msgHashing.size(),  // message
                          nullptr, 0,  // no context needed
                          g_publicKey  // pubkey
        );

    // Display results
    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT, "Raw data updated: %.*s",
                (int)rawDataString.length(), rawDataString.data());
    std::cout << "[Signature][Measure Part] Signature Receiving Time: "
              << signing_time << " current counter: " << measure_counter
              << std::endl;
    measure_counter++;

    std::cout << "Signature validity: ";

    if (0 == sigVerifyResult) {
        std::cout
            << "\033[1;32m";  // Color print (only in UNIX) -> BOLD + GREEN
        std::cout << "SIGNATURE VALID";
        std::cout << "\033[0m";  // back to normal Text
        std::cout << std::endl;
    } else {
        std::cout << "\033[1;31m";  // Color print (only in UNIX) -> BOLD + RED
        std::cout << "SIGNATURE INVALID";
        std::cout << "\033[0m";  // back to normal text
        std::cout << std::endl;
    }

    std::cout << "\n" << std::endl;
}

/**
 * @brief Setup subscriptions for both raw data and signature
 */
UA_StatusCode setupSubscriptions(UA_Client *client) {
    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                "Setting up subscriptions...");

    // Create subscription
    UA_CreateSubscriptionRequest request =
        UA_CreateSubscriptionRequest_default();
    request.requestedPublishingInterval = 1.0;  // 20ms publishing interval
    request.requestedMaxKeepAliveCount = 3;
    request.requestedLifetimeCount = 10;
    request.priority = 255;

    UA_CreateSubscriptionResponse response =
        UA_Client_Subscriptions_create(client, request, NULL, NULL, NULL);

    if (response.responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                     "Failed to create subscription: %s",
                     UA_StatusCode_name(response.responseHeader.serviceResult));
        return response.responseHeader.serviceResult;
    }

    UA_UInt32 subscriptionId = response.subscriptionId;
    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                "Subscription created with ID: %u", subscriptionId);

    // Setup monitored item for raw data
    UA_NodeId rawDataNodeId = UA_NODEID_STRING(
        1, const_cast<char *>("current.sensor.rawdata.string"));

    UA_MonitoredItemCreateRequest rawDataMonRequest =
        UA_MonitoredItemCreateRequest_default(rawDataNodeId);
    rawDataMonRequest.requestedParameters.samplingInterval =
        0.0;  // 50ms sampling
    rawDataMonRequest.requestedParameters.queueSize = 1;
    rawDataMonRequest.requestedParameters.discardOldest = true;

    UA_MonitoredItemCreateResult rawDataMonResponse =
        UA_Client_MonitoredItems_createDataChange(
            client, subscriptionId, UA_TIMESTAMPSTORETURN_BOTH,
            rawDataMonRequest, NULL, rawDataChangeCallback, NULL);

    if (rawDataMonResponse.statusCode != UA_STATUSCODE_GOOD) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                     "Failed to create raw data monitored item: %s",
                     UA_StatusCode_name(rawDataMonResponse.statusCode));
        return rawDataMonResponse.statusCode;
    }

    // Setup monitored item for signature
    UA_NodeId signatureNodeId =
        UA_NODEID_STRING(1, const_cast<char *>("current.sensor.signature"));

    UA_MonitoredItemCreateRequest sigMonRequest =
        UA_MonitoredItemCreateRequest_default(signatureNodeId);
    sigMonRequest.requestedParameters.samplingInterval = 0.0;  // 20ms sampling
    sigMonRequest.requestedParameters.queueSize = 1;
    sigMonRequest.requestedParameters.discardOldest = true;

    UA_MonitoredItemCreateResult sigMonResponse =
        UA_Client_MonitoredItems_createDataChange(
            client, subscriptionId, UA_TIMESTAMPSTORETURN_BOTH, sigMonRequest,
            NULL, signatureChangeCallback, NULL);

    if (sigMonResponse.statusCode != UA_STATUSCODE_GOOD) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                     "Failed to create signature monitored item: %s",
                     UA_StatusCode_name(sigMonResponse.statusCode));
        return sigMonResponse.statusCode;
    }

    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                "Successfully created monitored items (Raw: %u, Signature: %u)",
                rawDataMonResponse.monitoredItemId,
                sigMonResponse.monitoredItemId);

    return UA_STATUSCODE_GOOD;
}

int main(int argc, char *argv[]) {
    // Register the signals
    signal(SIGINT, stopHandler);
    signal(SIGTERM, stopHandler);

    // Define the server URL
    const char *serverUrl = "opc.tcp://10.45.0.2:4840";
    if (argc > 1) {
        serverUrl = argv[1];
    }

    // Create the client
    UA_Client *client = UA_Client_new();
    UA_ClientConfig *config = UA_Client_getConfig(client);

    UA_ClientConfig_setDefault(config);

    // set the send and buffer size as 64kb (same as default)
    // config->localConnectionConfig.sendBufferSize = 65536;
    // config->localConnectionConfig.recvBufferSize = 65536;

    // // set identical message size with the server
    // config->localConnectionConfig.localMaxMessageSize = 262144;
    // config->localConnectionConfig.remoteMaxMessageSize = 262144;

    // // set the same the configuration as the servers
    // config->localConnectionConfig.localMaxChunkCount = 8;
    // config->localConnectionConfig.remoteMaxChunkCount = 8;

    // Connect client to server
    UA_StatusCode status = UA_Client_connect(client, serverUrl);

    if (status != UA_STATUSCODE_GOOD) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                     "Cannot connect to the server %s - ErrorCode: %s\n",
                     serverUrl, UA_StatusCode_name(status));

        // Delete the client
        UA_Client_delete(client);

        // Exit main thread
        return EXIT_FAILURE;
    }

    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                "Connected to server: %s\n", serverUrl);

    // Initialize certificate and signature handler
    g_cert = new CertParser(
        "/home/vanbu/KeyAndCertificate/OpcUA/mldsa_65/certificate.der");
    g_sigHndl = new MLDSA<ParamSet_65>();

    // Certificate validation failure -> not allowing client to run
    if (false == g_cert->certIsValid()) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                     "Certificate validation failed. Exiting...");
        running = false;
    } else {
        // Copy public key to the array
        for (size_t i = 0; i < g_cert->publicKey.size(); i++) {
            g_publicKey.at(i) = g_cert->publicKey.at(i);
        }

        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_CLIENT,
                    "Certificate validated successfully");
    }

    // Setup subscriptions
    if (running) {
        status = setupSubscriptions(client);
        if (status != UA_STATUSCODE_GOOD) {
            running = false;
        }
    }

    // Main event loop
    while (running) {
        // Process OPC UA events (subscriptions)
        UA_Client_run_iterate(client, 1);

        // check if data is ready for response
        if (g_sensorData.isDataReady()) {
            processSensorData();
        }
    }

    // Cleanup
    UA_Client_disconnect(client);
    UA_Client_delete(client);

    delete g_cert;
    delete g_sigHndl;

    return EXIT_SUCCESS;
}