test_tls_integration.cpp File Reference

TLS Integration Tests - Secure DICOM Communication. More...

#include "test_fixtures.h"
#include <catch2/catch_test_macros.hpp>
#include <catch2/matchers/catch_matchers_string.hpp>
#include "kcenon/pacs/integration/network_adapter.h"
#include "kcenon/pacs/network/dimse/dimse_message.h"
#include "kcenon/pacs/services/verification_scp.h"
#include <filesystem>
#include <thread>

Include dependency graph for test_tls_integration.cpp:

Go to the source code of this file.

Functions
	TEST_CASE ("TLS C-ECHO connection", "[tls][connectivity]")
	TEST_CASE ("TLS certificate validation", "[tls][security]")
	TEST_CASE ("Mutual TLS authentication", "[tls][mtls]")
	TEST_CASE ("TLS version negotiation", "[tls][version]")
	TEST_CASE ("Multiple concurrent TLS connections", "[tls][concurrent]")
	TEST_CASE ("TLS configuration validation", "[tls][config]")

Detailed Description

TLS Integration Tests - Secure DICOM Communication.

Tests TLS-secured DICOM communication using the secure_dicom example applications and network_adapter TLS configuration.

See also

Issue #141 - TLS Integration Tests

DICOM PS3.15 - Security and System Management Profiles

Test Scenarios:

1. Basic TLS connection
2. Certificate validation
3. Mutual TLS (mTLS)
4. TLS-secured store/query workflow

Definition in file test_tls_integration.cpp.

Function Documentation

TEST_CASE() [1/6]

TEST_CASE	(	"Multiple concurrent TLS connections"	,
		""	[tls][concurrent] )

Definition at line 598 of file test_tls_integration.cpp.

                                                                      {
    auto certs = get_test_certificates();
 
    if (!certs.all_exist()) {
        SKIP("Test certificates not available");
    }
 
    tls_config server_tls;
    server_tls.enabled = true;
    server_tls.cert_path = certs.server_cert;
    server_tls.key_path = certs.server_key;
    server_tls.ca_path = certs.ca_cert;
    server_tls.verify_peer = false;
    server_tls.min_version = tls_config::tls_version::v1_2;
 
    auto port = find_available_port();
    tls_test_server server(port, "CONCURRENT_TLS", server_tls);
 
    if (!server.is_tls_valid()) {
        SKIP("TLS not properly configured");
    }
 
    server.register_service(std::make_shared<verification_scp>());
    REQUIRE(server.start());
 
    constexpr int num_connections = 3;
    std::vector<std::thread> threads;
    std::atomic<int> success_count{0};
 
    for (int i = 0; i < num_connections; ++i) {
        threads.emplace_back([&, i]() {
            tls_config client_tls;
            client_tls.enabled = true;
            client_tls.ca_path = certs.ca_cert;
            client_tls.verify_peer = true;
            client_tls.min_version = tls_config::tls_version::v1_2;
 
            auto result = tls_test_client::connect(
                "localhost", port, server.ae_title(),
                "TLS_SCU_" + std::to_string(i), client_tls);
 
            if (result.is_err()) {
                return;
            }
 
            auto& assoc = result.value();
            auto ctx_opt = assoc.accepted_context_id(verification_sop_class_uid);
            if (!ctx_opt) {
                return;
            }
 
            using namespace kcenon::pacs::network::dimse;
            auto echo_rq = make_c_echo_rq(1, verification_sop_class_uid);
            auto send_result = assoc.send_dimse(*ctx_opt, echo_rq);
            if (send_result.is_err()) {
                return;
            }
 
            auto recv_result = assoc.receive_dimse(default_timeout());
            if (recv_result.is_ok()) {
                auto& [ctx, rsp] = recv_result.value();
                if (rsp.status() == status_success) {
                    ++success_count;
                }
            }
 
            (void)assoc.release(default_timeout());
        });
    }
 
    for (auto& t : threads) {
        t.join();
    }
 
    server.stop();
 
    REQUIRE(success_count == num_connections);
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration::tls_config::key_path, kcenon::pacs::integration::tls_config::min_version, kcenon::pacs::services::verification_sop_class_uid, and kcenon::pacs::integration::tls_config::verify_peer.

Here is the call graph for this function:

TEST_CASE() [2/6]

TEST_CASE	(	"Mutual TLS authentication"	,
		""	[tls][mtls] )

Definition at line 406 of file test_tls_integration.cpp.

                                                      {
    auto certs = get_test_certificates();
 
    if (!certs.all_exist()) {
        SKIP("Test certificates not available");
    }
 
    SECTION("Client with valid certificate succeeds") {
        // Server requires client certificate
        tls_config server_tls;
        server_tls.enabled = true;
        server_tls.cert_path = certs.server_cert;
        server_tls.key_path = certs.server_key;
        server_tls.ca_path = certs.ca_cert;
        server_tls.verify_peer = true;  // Require client cert
        server_tls.min_version = tls_config::tls_version::v1_2;
 
        auto port = find_available_port();
        tls_test_server server(port, "MTLS_SCP", server_tls);
 
        if (!server.is_tls_valid()) {
            SKIP("TLS not properly configured");
        }
 
        server.register_service(std::make_shared<verification_scp>());
        REQUIRE(server.start());
 
        // Client with certificate
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.cert_path = certs.client_cert;
        client_tls.key_path = certs.client_key;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "MTLS_SCU", client_tls);
 
        REQUIRE(result.is_ok());
        auto& assoc = result.value();
 
        // Verify we can communicate
        REQUIRE(assoc.has_accepted_context(verification_sop_class_uid));
 
        using namespace kcenon::pacs::network::dimse;
        auto ctx_id = *assoc.accepted_context_id(verification_sop_class_uid);
        auto echo_rq = make_c_echo_rq(1, verification_sop_class_uid);
        auto send_result = assoc.send_dimse(ctx_id, echo_rq);
        REQUIRE(send_result.is_ok());
 
        auto recv_result = assoc.receive_dimse(default_timeout());
        REQUIRE(recv_result.is_ok());
 
        auto& [recv_ctx, rsp] = recv_result.value();
        REQUIRE(rsp.status() == status_success);
 
        (void)assoc.release(default_timeout());
        server.stop();
    }
 
    SECTION("Client without certificate fails when server requires it") {
        tls_config server_tls;
        server_tls.enabled = true;
        server_tls.cert_path = certs.server_cert;
        server_tls.key_path = certs.server_key;
        server_tls.ca_path = certs.ca_cert;
        server_tls.verify_peer = true;  // Require client cert
        server_tls.min_version = tls_config::tls_version::v1_2;
 
        auto port = find_available_port();
        tls_test_server server(port, "MTLS_SCP", server_tls);
 
        if (!server.is_tls_valid()) {
            SKIP("TLS not properly configured");
        }
 
        server.register_service(std::make_shared<verification_scp>());
        REQUIRE(server.start());
 
        // Client without certificate
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;
        // No client cert/key
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "NO_CERT_SCU", client_tls);
 
        // Should fail due to missing client certificate
        // Note: Actual behavior depends on server configuration
        if (result.is_ok()) {
            auto& assoc = result.value();
            // Connection might succeed but operations may fail
            assoc.abort();
        }
 
        server.stop();
    }
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration::tls_config::key_path, kcenon::pacs::integration::tls_config::min_version, kcenon::pacs::services::verification_sop_class_uid, and kcenon::pacs::integration::tls_config::verify_peer.

Here is the call graph for this function:

TEST_CASE() [3/6]

TEST_CASE	(	"TLS C-ECHO connection"	,
		""	[tls][connectivity] )

Definition at line 247 of file test_tls_integration.cpp.

                                                          {
    auto certs = get_test_certificates();
 
    // Skip if certificates not available
    if (!certs.all_exist()) {
        WARN("Skipping TLS tests: certificates not found at " << certs.ca_cert.parent_path());
        SKIP("Test certificates not available");
    }
 
    SECTION("TLS server accepts connection and responds to C-ECHO") {
        // Configure TLS server
        tls_config server_tls;
        server_tls.enabled = true;
        server_tls.cert_path = certs.server_cert;
        server_tls.key_path = certs.server_key;
        server_tls.ca_path = certs.ca_cert;
        server_tls.verify_peer = false;  // Don't require client cert
        server_tls.min_version = tls_config::tls_version::v1_2;
 
        auto port = find_available_port();
        tls_test_server server(port, "TLS_SCP", server_tls);
 
        if (!server.is_tls_valid()) {
            WARN("TLS configuration not valid, skipping test");
            SKIP("TLS not properly configured");
        }
 
        server.register_service(std::make_shared<verification_scp>());
        REQUIRE(server.start());
        REQUIRE(server.is_running());
 
        // Configure TLS client
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;  // Verify server cert
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto connect_result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "TLS_SCU", client_tls);
 
        REQUIRE(connect_result.is_ok());
        auto& assoc = connect_result.value();
 
        // Verify connection
        REQUIRE(assoc.has_accepted_context(verification_sop_class_uid));
 
        auto context_id_opt = assoc.accepted_context_id(verification_sop_class_uid);
        REQUIRE(context_id_opt.has_value());
 
        // Send C-ECHO
        using namespace kcenon::pacs::network::dimse;
        auto echo_rq = make_c_echo_rq(1, verification_sop_class_uid);
        auto send_result = assoc.send_dimse(*context_id_opt, echo_rq);
        REQUIRE(send_result.is_ok());
 
        auto recv_result = assoc.receive_dimse(default_timeout());
        REQUIRE(recv_result.is_ok());
 
        auto& [recv_ctx, echo_rsp] = recv_result.value();
        REQUIRE(echo_rsp.command() == command_field::c_echo_rsp);
        REQUIRE(echo_rsp.status() == status_success);
 
        // Clean up
        (void)assoc.release(default_timeout());
        server.stop();
    }
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration::tls_config::key_path, kcenon::pacs::integration::tls_config::min_version, kcenon::pacs::services::verification_sop_class_uid, and kcenon::pacs::integration::tls_config::verify_peer.

Here is the call graph for this function:

TEST_CASE() [4/6]

TEST_CASE	(	"TLS certificate validation"	,
		""	[tls][security] )

Definition at line 320 of file test_tls_integration.cpp.

                                                           {
    auto certs = get_test_certificates();
 
    if (!certs.all_exist()) {
        SKIP("Test certificates not available");
    }
 
    // Start TLS server
    tls_config server_tls;
    server_tls.enabled = true;
    server_tls.cert_path = certs.server_cert;
    server_tls.key_path = certs.server_key;
    server_tls.ca_path = certs.ca_cert;
    server_tls.verify_peer = false;
    server_tls.min_version = tls_config::tls_version::v1_2;
 
    auto port = find_available_port();
    tls_test_server server(port, "TLS_SCP", server_tls);
 
    if (!server.is_tls_valid()) {
        SKIP("TLS not properly configured");
    }
 
    server.register_service(std::make_shared<verification_scp>());
    REQUIRE(server.start());
 
    SECTION("Valid certificate succeeds") {
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "TLS_SCU", client_tls);
 
        REQUIRE(result.is_ok());
        auto& assoc = result.value();
        (void)assoc.release(default_timeout());
    }
 
    SECTION("Wrong CA fails validation") {
        // Skip if other_ca doesn't exist
        if (!std::filesystem::exists(certs.other_ca_cert)) {
            WARN("other_ca.crt not found, skipping wrong CA test");
            server.stop();
            return;
        }
 
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.other_ca_cert;  // Wrong CA
        client_tls.verify_peer = true;
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "TLS_SCU", client_tls);
 
        // Should fail due to certificate verification failure
        // Note: Actual behavior depends on implementation
        // Some implementations may still connect but with warnings
        if (result.is_ok()) {
            auto& assoc = result.value();
            assoc.abort();
        }
    }
 
    SECTION("TLS configuration validation") {
        // Test invalid TLS configuration
        tls_config invalid_tls;
        invalid_tls.enabled = true;
        invalid_tls.cert_path = "/nonexistent/cert.pem";
        invalid_tls.key_path = "/nonexistent/key.pem";
 
        auto result = network_adapter::configure_tls(invalid_tls);
        // Should fail validation
        REQUIRE(result.is_err());
    }
 
    server.stop();
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration::tls_config::key_path, kcenon::pacs::integration::tls_config::min_version, and kcenon::pacs::integration::tls_config::verify_peer.

Here is the call graph for this function:

TEST_CASE() [5/6]

TEST_CASE	(	"TLS configuration validation"	,
		""	[tls][config] )

Definition at line 681 of file test_tls_integration.cpp.

                                                           {
    SECTION("Disabled TLS is always valid") {
        tls_config cfg;
        cfg.enabled = false;
 
        REQUIRE(cfg.is_valid() == true);
    }
 
    SECTION("Enabled TLS requires cert and key") {
        tls_config cfg;
        cfg.enabled = true;
 
        // Missing both cert and key
        REQUIRE(cfg.is_valid() == false);
 
        // Only cert
        cfg.cert_path = "/some/cert.pem";
        REQUIRE(cfg.is_valid() == false);
 
        // Both cert and key
        cfg.key_path = "/some/key.pem";
        REQUIRE(cfg.is_valid() == true);
    }
 
    SECTION("CA path is optional") {
        tls_config cfg;
        cfg.enabled = true;
        cfg.cert_path = "/some/cert.pem";
        cfg.key_path = "/some/key.pem";
 
        // Valid without CA
        REQUIRE(cfg.is_valid() == true);
 
        // Still valid with CA
        cfg.ca_path = "/some/ca.pem";
        REQUIRE(cfg.is_valid() == true);
    }
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration::tls_config::is_valid(), and kcenon::pacs::integration::tls_config::key_path.

Here is the call graph for this function:

TEST_CASE() [6/6]

TEST_CASE	(	"TLS version negotiation"	,
		""	[tls][version] )

Definition at line 513 of file test_tls_integration.cpp.

                                                       {
    auto certs = get_test_certificates();
 
    if (!certs.all_exist()) {
        SKIP("Test certificates not available");
    }
 
    SECTION("TLS 1.2 connection") {
        tls_config server_tls;
        server_tls.enabled = true;
        server_tls.cert_path = certs.server_cert;
        server_tls.key_path = certs.server_key;
        server_tls.ca_path = certs.ca_cert;
        server_tls.verify_peer = false;
        server_tls.min_version = tls_config::tls_version::v1_2;
 
        auto port = find_available_port();
        tls_test_server server(port, "TLS12_SCP", server_tls);
 
        if (!server.is_tls_valid()) {
            SKIP("TLS not properly configured");
        }
 
        server.register_service(std::make_shared<verification_scp>());
        REQUIRE(server.start());
 
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;
        client_tls.min_version = tls_config::tls_version::v1_2;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "TLS12_SCU", client_tls);
 
        REQUIRE(result.is_ok());
        auto& assoc = result.value();
        (void)assoc.release(default_timeout());
        server.stop();
    }
 
    SECTION("TLS 1.3 connection") {
        tls_config server_tls;
        server_tls.enabled = true;
        server_tls.cert_path = certs.server_cert;
        server_tls.key_path = certs.server_key;
        server_tls.ca_path = certs.ca_cert;
        server_tls.verify_peer = false;
        server_tls.min_version = tls_config::tls_version::v1_3;
 
        auto port = find_available_port();
        tls_test_server server(port, "TLS13_SCP", server_tls);
 
        if (!server.is_tls_valid()) {
            SKIP("TLS not properly configured");
        }
 
        server.register_service(std::make_shared<verification_scp>());
        REQUIRE(server.start());
 
        tls_config client_tls;
        client_tls.enabled = true;
        client_tls.ca_path = certs.ca_cert;
        client_tls.verify_peer = true;
        client_tls.min_version = tls_config::tls_version::v1_3;
 
        auto result = tls_test_client::connect(
            "localhost", port, server.ae_title(), "TLS13_SCU", client_tls);
 
        // TLS 1.3 may not be supported on all systems
        if (result.is_ok()) {
            auto& assoc = result.value();
            (void)assoc.release(default_timeout());
        } else {
            INFO("TLS 1.3 not supported: " << result.error().message);
        }
 
        server.stop();
    }
}

References kcenon::pacs::integration::tls_config::ca_path, kcenon::pacs::integration::tls_config::cert_path, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration::tls_config::enabled, kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration::tls_config::key_path, kcenon::pacs::integration::tls_config::min_version, and kcenon::pacs::integration::tls_config::verify_peer.

Here is the call graph for this function: