iron_house_server.py

#!/usr/bin/env python

'''
Ironhouse extends Stonehouse with client public key authentication.

This is the strongest security model we have today, protecting against every
attack we know about, except end-point attacks (where an attacker plants
spyware on a machine to capture data before it's encrypted, or after it's
decrypted).

Author: Chris Laws

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

To run from an erlang shell:
cd("python-test"),
{ok, ServerKeys} = chumak_cert:read("server.key"),
SPK = proplists:get_value(public_key, ServerKeys),
{ok, ClientKeys} = chumak_cert:read("client.key"),
CSK = proplists:get_value(secret_key, ClientKeys),
CPK = proplists:get_value(public_key, ClientKeys),
application:start(chumak),
{ok, Socket} = chumak:socket(pull),
ok = chumak:set_socket_option(Socket, curve_secretkey, CSK),
ok = chumak:set_socket_option(Socket, curve_publickey, CPK),
ok = chumak:set_socket_option(Socket, curve_serverkey, SPK),
{ok, _} = chumak:connect(Socket, tcp, "127.0.0.1", 9000),
{ok, Message} = chumak:recv(Socket),
io:format("received: ~p~n", [Message]),
halt().

'''

import logging
import os
import sys
import time

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


def run():
    ''' Run Ironhouse 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 authenticator to use the certificate in a directory
    print(keys_dir)
    #auth.configure_curve(domain='*', location=keys_dir)
    auth.configure_curve(domain='*', location=".")

    server_key_file = os.path.join(keys_dir, "server.key")
    server_public, server_secret = zmq.auth.load_certificate(server_key_file)

    server = ctx.socket(zmq.PUSH)
    server.curve_secretkey = server_secret
    server.curve_publickey = server_public
    server.curve_server = True  # must come before bind
    server.bind('tcp://*:9000')

    server.send(b"Hello")
    # Make sure that there is time to finish the handshake
    time.sleep(2)

    # 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()