08-ssl-bridge-bidirectional.py

#!/usr/bin/env python

import time
import subprocess
import socket
import ssl

import inspect, os, sys
# From http://stackoverflow.com/questions/279237/python-import-a-module-from-a-folder
cmd_subfolder = os.path.realpath(os.path.abspath(os.path.join(os.path.split(inspect.getfile( inspect.currentframe() ))[0],"..")))
if cmd_subfolder not in sys.path:
    sys.path.insert(0, cmd_subfolder)

import mosq_test

def start_docker_broker(filename, cmd=None, port=0, use_conf=False):
    delay = 1

    if use_conf == True:
        cmd = ['docker', 'run', '--rm', '-w', '/etc/mosquitto/broker', '-v', os.getcwd() + '/..:/etc/mosquitto', '-p', str(port) + ':' + str(port), 'eclipse-mosquitto', 'mosquitto', '-v', '-c', filename.replace('.py', '.conf')]

        if port == 0:
            port = 1888
    else:
        if cmd is None and port != 0:
            cmd = ['docker', 'run', '--rm', '-w', '/etc/mosquitto/broker', '-v', os.getcwd() + '/..:/etc/mosquitto', '-p', str(port) + ':' + str(port), 'eclipse-mosquitto', 'mosquitto', '-v', '-p', str(port)]
        elif cmd is None and port == 0:
            port = 1888
            cmd = ['docker', 'run', '--rm', '-w', '/etc/mosquitto/broker', '-v', os.getcwd() + '/..:/etc/mosquitto', '-p', str(port) + ':' + str(port), 'eclipse-mosquitto', 'mosquitto', '-v', '-c', filename.replace('.py', '.conf')]
        elif cmd is not None and port == 0:
            port = 1888
            
    print(port)
    print(cmd)
    broker = subprocess.Popen(cmd, stderr=subprocess.PIPE)
    for i in range(0, 20):
        time.sleep(delay)
        c = None
        try:
            c = socket.create_connection(("localhost", port))
        except socket.error as err:
            if err.errno != errno.ECONNREFUSED:
                raise

        if c is not None:
            c.close()
            time.sleep(delay)
            return broker
    raise IOError

def do_client_connect(connect_packet, connack_packet, hostname="localhost", port=1888, timeout=60, connack_error="connack", sock=False):
    if not sock:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    sock.connect((hostname, port))

    return mosq_test.do_send_receive(sock, connect_packet, connack_packet, connack_error)



def write_far_config(filename, portfar):
    with open(filename, 'w') as f:
        f.write("port %d\n" % (portfar))
        f.write("\n")
        f.write("protocol mqtt\n")
        f.write("tls_version tlsv1.2\n")
        f.write("cafile ../ssl/all-ca.crt\n")
        f.write("certfile ../ssl/server.crt\n")
        f.write("keyfile ../ssl/server.key\n")
        f.write("require_certificate true\n")
        f.write("use_identity_as_username true\n")
        f.write("crlfile ../ssl/crl.pem\n")
        f.write("ciphers HIGH\n")
        f.write("connection_messages true\n")
        f.write("log_dest stdout\n")
        f.write("log_timestamp true\n")
        f.write("use_username_as_clientid true\n")
        f.write("allow_anonymous false\n")
        f.write("acl_file user.acl\n")
        f.write("autosave_interval 1\n")
        f.write("autosave_on_changes true\n")

#mount_point controllers
##persistence true
##persistence_file mosquitto.db
##persistence_location /db/


def write_near_config(filename, portfar, portnear):
    with open(filename, 'w') as f:
        f.write("port %d\n" % (portnear))
        f.write("\n")
        f.write("connection bridge_test\n")
        f.write("address 127.0.0.1:%d\n" % (portfar))
        f.write("topic bridge/# both 0\n")
        f.write("notifications false\n")
        f.write("restart_timeout 2\n")
        f.write("\n")
        f.write("bridge_cafile ../ssl/all-ca.crt\n")
        f.write("bridge_certfile ../ssl/bridge.crt\n")
        f.write("bridge_keyfile ../ssl/bridge.key\n")
        f.write("bridge_tls_version tlsv1.2\n")
        f.write("bridge_insecure true\n")
        f.write("log_dest stdout\n")

(portfar, portnear) = mosq_test.get_port(2)
far_conf_file = os.path.basename(__file__).replace('.py', 'far.conf')
write_far_config(far_conf_file, portfar)
near_conf_file = os.path.basename(__file__).replace('.py', 'near.conf')
write_near_config(near_conf_file, portfar, portnear)

rc = 1
keepalive = 60
client_id = socket.gethostname()+".bridge_test"
connect_packet = mosq_test.gen_connect(client_id, keepalive=keepalive, clean_session=False, proto_ver=128+4)
connack_packet = mosq_test.gen_connack(rc=0)

mid = 1
subscribe_packet = mosq_test.gen_subscribe(mid, "bridge/#", 0)
suback_packet = mosq_test.gen_suback(mid, 0)

publish_packet = mosq_test.gen_publish("bridge/ssl/test", qos=0, payload="message")

far_broker = start_docker_broker(filename=far_conf_file, port=portfar, use_conf=True)
near_broker = mosq_test.start_broker(filename=near_conf_file, port=portnear, use_conf=True)

far_connect_packet = mosq_test.gen_connect("test-far-client", keepalive=keepalive)
far_connack_packet = mosq_test.gen_connack(rc=0)
far_subscribe_packet = mosq_test.gen_subscribe(1, "bridge/ssl/testnear", qos=0)
far_publish_packet = mosq_test.gen_publish("bridge/ssl/testfar", qos=0, payload="farmessage")
far_disconnect_packet = mosq_test.gen_disconnect()

near_connect_packet = mosq_test.gen_connect("test-near-client", keepalive=keepalive)
near_connack_packet = mosq_test.gen_connack(rc=0)
near_subscribe_packet = mosq_test.gen_subscribe(1, "bridge/ssl/testfar", qos=0)
near_publish_packet = mosq_test.gen_publish("bridge/ssl/testnear", qos=0, payload="nearmessage")
near_disconnect_packet = mosq_test.gen_disconnect()

suback_packet = mosq_test.gen_suback(mid, 0)

sfar_sock = False
near_sock = False

try:
    fsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    fsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    fsockssl = ssl.wrap_socket(fsock, ca_certs="../ssl/all-ca.crt", keyfile="../ssl/client.key", certfile="../ssl/client.crt", server_side=False, ssl_version=ssl.PROTOCOL_TLSv1_2)
    sfar_sock = do_client_connect(far_connect_packet, far_connack_packet, port=portfar, connack_error="far client connack", sock=fsockssl)
    near_sock = mosq_test.do_client_connect(near_connect_packet, near_connack_packet, port=portnear, connack_error="near client connack")
    sfar_sock.send(far_subscribe_packet)
    near_sock.send(near_subscribe_packet)

    sfar_sock.send(far_publish_packet)
    near_sock.send(near_publish_packet)

    mosq_test.expect_packet(near_sock, "suback near", suback_packet)
    mosq_test.expect_packet(sfar_sock, "suback far", suback_packet)

    if mosq_test.expect_packet(near_sock, "far_receive", far_publish_packet):

        if mosq_test.expect_packet(sfar_sock, "near_recieve", near_publish_packet):
        
            sfar_sock.send(far_disconnect_packet)
            sfar_sock.close()
            near_sock.send(near_disconnect_packet)
            near_sock.close()

finally:
    os.remove(near_conf_file)
    os.remove(far_conf_file)
    try:
        if sfar_sock:
          sfar_sock.close()
    except NameError:
        pass
    try:
        if near_sock:
            near_sock.close()
    except NameError:
        pass

    far_broker.terminate()
    near_broker.terminate()
    far_broker.wait()
    near_broker.wait()
    (stdo, stde) = far_broker.communicate()
    if rc:
        print(stde)
    (stdo, stde) = near_broker.communicate()
    if rc:
        print(stde)
    if near_sock:
        near_sock.close()
    if sfar_sock:
        sfar_sock.close()

exit(rc)