stone_house_client.py

#!/usr/bin/env python

'''
Stonehouse uses the "CURVE" security mechanism.
This gives us strong encryption on data, and (as far as we know) unbreakable
authentication. Stonehouse is the minimum you would use over public networks,
and assures clients that they are speaking to an authentic server, while
allowing any client to connect.
Author: Chris Laws

Modified by Willem de Jong - only start the Python client, the server is the
Chumak Erlang implementation.

To run, start an Erlang shell and issue the following commands:
cd("python-test"),
{ok, ServerKeys} = chumak_cert:read("server.key"),
SK = proplists:get_value(secret_key, ServerKeys),
application:start(chumak),
{ok, Socket} = chumak:socket(push),
ok = chumak:set_socket_option(Socket, curve_server, true),
ok = chumak:set_socket_option(Socket, curve_secretkey, SK),
{ok, _BindProc} = chumak:bind(Socket, tcp, "127.0.0.1", 9000).
timer:sleep(1000),
chumak:send(Socket, <<"Hello">>),
halt().

'''

import logging
import os
import sys
import time

import zmq
import zmq.auth
from zmq.auth.thread import ThreadAuthenticator


def run():
    ''' Run Stonehouse example '''

    # These directories are generated by the generate_certificates script
    keys_dir = os.path.dirname(__file__)

    ctx = zmq.Context.instance()

    # Start an authenticator for this context.
    auth = ThreadAuthenticator(ctx)
    auth.start()
    auth.allow('127.0.0.1')
    # Tell the authenticator how to handle CURVE requests
    auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY)

    client = ctx.socket(zmq.PULL)
    # We need two certificates, one for the client and one for
    # the server. The client must know the server's public key
    # to make a CURVE connection.
    client_secret_file = os.path.join(keys_dir, "client.key")
    client_public, client_secret = zmq.auth.load_certificate(client_secret_file)
    client.curve_secretkey = client_secret
    client.curve_publickey = client_public

    # The client must know the server's public key to make a CURVE connection.
    server_public_file = os.path.join(keys_dir, "server.key")
    server_public, _ = zmq.auth.load_certificate(server_public_file)
    client.curve_serverkey = server_public

    client.connect('tcp://127.0.0.1:9000')

    if client.poll(100000):
        msg = client.recv()
        if msg == b"Hello":
            logging.info("Stonehouse test OK")
    else:
        logging.error("Stonehouse test FAIL")

    # stop auth thread
    auth.stop()

if __name__ == '__main__':
    if zmq.zmq_version_info() < (4,0):
        raise RuntimeError("Security is not supported in libzmq version < 4.0. libzmq version {0}".format(zmq.zmq_version()))

    if '-v' in sys.argv:
        level = logging.DEBUG
    else:
        level = logging.INFO

    logging.basicConfig(level=level, format="[%(levelname)s] %(message)s")

    run()