solace-samples-python/howtos/how_to_configure_authentication.py at main · SolaceSamples/solace-samples-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""sampler module for configuring the authentication"""
from string import Template

from solace.messaging.config.authentication_strategy import BasicUserNamePassword, ClientCertificateAuthentication
from solace.messaging.config.solace_properties import authentication_properties
from solace.messaging.config.transport_security_strategy import TLS
from solace.messaging.messaging_service import MessagingService
from howtos.sampler_boot import SamplerBoot, SamplerUtil, SolaceConstants

boot = SamplerBoot()


class HowToConfigureAuthentication:
    """class contains methods for configuring the authentication"""

    @staticmethod
    def configure_basic_auth_credentials(props, user_name: str, password: str):
        """setup for basic auth using user name and password

        Args:
            props:
            user_name: user name
            password: password

        Returns:
            configured and connected instance of MessagingService ready to be used for messaging tasks
        """
        try:
            messaging_service = MessagingService.builder().from_properties(props) \
                .with_authentication_strategy(BasicUserNamePassword.of(user_name, password)).build()
            return messaging_service.connect()
        except Exception as exception:
            print(exception)
        finally:
            messaging_service.disconnect()

    @staticmethod
    def configure_client_certificate_authentication_customized_settings(props, key_file,
                                                                        key_store_password, key_store_url):
        """
        For a client to use a client certificate authentication scheme, the host event broker must be
        properly configured for TLS/SSL connections, and Client Certificate Verification must be
        enabled for the particular Message VPN that the client is connecting to. On client side client
        certificate needs to be present in a keystore file.

        Args:
            props:
            key_store_password: password for the key store
            key_store_url: url to the key store file
            key_file: key file

        Returns:
            configured and connected instance of MessagingService ready to be used for messaging tasks
        """
        try:
            transport_security = TLS.create() \
                .with_certificate_validation(True, validate_server_name=False,
                                             trust_store_file_path=SamplerUtil.get_trusted_store_dir())

            messaging_service = MessagingService.builder() \
                .from_properties(props) \
                .with_transport_security_strategy(transport_security) \
                .with_authentication_strategy(ClientCertificateAuthentication.of(certificate_file=key_store_url,
                                                                                 key_file=key_file,
                                                                                 key_password=key_store_password)) \
                .build(SamplerUtil.get_new_application_id())
            return messaging_service.connect()
        except Exception as exception:
            print(exception)
        finally:
            messaging_service.disconnect()

    @staticmethod
    def basic_compression(props, compression_range):
        """method for applying compression to the messaging service

        Args:
            props: broker properties
            compression_range: int value the compression value
        """
        try:
            messaging_service = MessagingService.builder().from_properties(props) \
                .with_message_compression(compression_range).build()
            return messaging_service.connect()
        finally:
            messaging_service.disconnect()

    @staticmethod
    def run():
        """method to run all the other authentication configuration methods"""
        props_unsecured = boot.broker_properties()
        props_secured = boot.secured_broker_properties()
        props_compressed = boot.compressed_broker_properties()
        user_name = props_unsecured[authentication_properties.SCHEME_BASIC_USER_NAME]
        password = props_unsecured[authentication_properties.SCHEME_BASIC_PASSWORD]
        key_store_url = SamplerUtil.get_valid_client_certificate()
        key_store_password = SolaceConstants.KEY_STORE_PASSWORD
        key_file = SamplerUtil.get_valid_client_key()

        result = HowToConfigureAuthentication.configure_basic_auth_credentials(props_unsecured, user_name, password)
        SamplerUtil.print_sampler_result(Template("Message Service[SYNC] connect with AUTH strategy $status")
                                         .substitute(status="SUCCESS" if result else "FAILED"))

        result = HowToConfigureAuthentication \
            .configure_client_certificate_authentication_customized_settings(props_secured, key_file,
                                                                             key_store_password, key_store_url)
        SamplerUtil.print_sampler_result(Template("Message Service[SYNC] connect with TLS strategy $status")
                                         .substitute(status="SUCCESS" if result else "FAILED"))

        result = HowToConfigureAuthentication.basic_compression(props_compressed, compression_range=1)
        SamplerUtil.print_sampler_result(Template("Message Service[SYNC] connect with COMPRESSION $status")
                                         .substitute(status="SUCCESS" if result else "FAILED"))


if __name__ == '__main__':
    HowToConfigureAuthentication.run()