tail client over stunnel

config.ini

[authentication]
redis_host = logging.utils.globuscs.info
certfile = /home/user/.tailclient/username.crt
keyfile = /home/user/.tailclient/username.key

install.py

#/usr/bin/env python
import os.path
import os
import sys
import ConfigParser
import readline
import glob
import urllib2

path = os.path.expanduser('~/.tailclient')

if not os.path.exists(path):
    os.mkdir(path)

if not os.path.isdir(path):
    print("~/.tailclient exists but is not a directory")
    sys.exit(1)


def complete(text, state):
    return (glob.glob(os.path.expanduser(text) + '*') + [None])[state]

readline.set_completer_delims(' \t\n;')
readline.parse_and_bind("tab: complete")
readline.set_completer(complete)

key = raw_input("Where is your keyfile? ")
cert = raw_input("Where is your certificate? ")
parser = ConfigParser.ConfigParser()
parser.add_section('authentication')
parser.set('authentication', 'redis_host', 'logging.utils.globuscs.info')
parser.set('authentication', 'keyfile', key)
parser.set('authentication', 'certfile', cert)
with open(os.path.join(path, 'config.ini'), 'w') as f:
    parser.write(f)

script_path = os.path.expanduser(raw_input("Where do you want to store the script? "))

script_dir = os.path.dirname(script_path)

if not os.path.exists(script_dir):
    os.makedirs(script_dir)

if not os.path.isdir(script_dir):
    print "{0} is not a directory".format(script_dir)

response = urllib2.urlopen('https://raw.github.com/gist/1827895/7cd89847f96f516d11a5b1e6c86e9ce011858f32/tailclient.py')
with open(script_path, 'wb') as f:
    f.write(response.read())

response = urllib2.urlopen('https://raw.github.com/gist/1827895/85a8bada1c2727e2cd92959dac187ea9b4ca3794/logstash.pem')
with open(os.path.expanduser('~/.tailclient/logstash.pem'), 'wb') as f:
    f.write(response.read())

logstash.pem

-----BEGIN CERTIFICATE-----
MIIF3TCCA8WgAwIBAgIJANeZbNwr8w2XMA0GCSqGSIb3DQEBBQUAMIGEMQswCQYD
VQQGEwJVUzERMA8GA1UECAwISWxsaW5vaXMxEDAOBgNVBAcMB0NoaWNhZ28xDzAN
BgNVBAoMBkdsb2J1czEZMBcGA1UECwwQZ2xvYnVzb25saW5lLm9yZzEkMCIGA1UE
AwwbRHVtbXkgQ2VydGlmaWNhdGUgQXV0aG9yaXR5MB4XDTEyMDIxNDE1MjMyMloX
DTEzMDIxMzE1MjMyMlowgYQxCzAJBgNVBAYTAlVTMREwDwYDVQQIDAhJbGxpbm9p
czEQMA4GA1UEBwwHQ2hpY2FnbzEPMA0GA1UECgwGR2xvYnVzMRkwFwYDVQQLDBBn
bG9idXNvbmxpbmUub3JnMSQwIgYDVQQDDBtEdW1teSBDZXJ0aWZpY2F0ZSBBdXRo
b3JpdHkwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQDiTV6aXMNyS+G2
HKQQJgN5cn7crKBSSMPj23/yqgSmkok499Y7jX2rrZQd1Wsaxcn1HFQAaStGihdg
fA6skht2nBC9hiDD7uv2CpaCFj+6tGhkApXo7y5ktpOt6JAVJa5AJ6JWDk4zZ3Fj
7Gq+V0RoKZ+n8AX8AfVTRZ62xhixILcbn3U4Rr0VwETa/CMyPbYSNL2nJe87lirz
vZVtzlc0987WfMbQrR9txfsfRdzQsNFXBGogXZo0FdXKAE1SZH7Pzg+HOfPIt+44
SL4sTIDBZWscj/bsMCSxvCph5ISqXW4sqjGq9RY8MfDOAj+Ab0vlBdOeCP65xvFZ
T3tTTmL3wi+9uvr9OoTQ3Llf4DTf2BN+oWD2aLBYk3wAtqoxkD7QJ2HySJMYzfXG
uPKaRJi3RI5c7ELm7LEX0lzJ4RntMilewkpUQxcOxx1NJTiqsdb939dIUfRqgxH7
Rc2FWEd7ZZsPaKB9A+/Vc1KNxpdcyeRT/4vRDXxdrYZUVRwQ6CuVrCa5NL4KWKax
0JzdlE1pcybRFeP9LDsCjBU2wxyW8snsmXFcHnf3o3b3MaG8Z1fnApJCuSQg6/UG
wETEjJdN8Q1lgSfW+iOYi6zYltiBa4KaAbvBx762j9D+ERJ0GlCnpP943ej2Q32l
bDW6ZUNDWvfKbhCUGQ+ZBa5u/qi7awIDAQABo1AwTjAdBgNVHQ4EFgQUR+qEzL02
3wgYAScSJni3/402xWswHwYDVR0jBBgwFoAUR+qEzL023wgYAScSJni3/402xWsw
DAYDVR0TBAUwAwEB/zANBgkqhkiG9w0BAQUFAAOCAgEAZQtqwlV9A6CCEe3z6i/X
h/4MFAJ+WOAt7yja4ydeSvQK5UGv+aY2l/pybQJxZ0Slydhu0sOeTnPO38LvmPSO
9hKhg/6Mw9FSuj1+NYd0AuY6KZkjedGbe4sgzz/wHfNQNqjqA3Y85t3fYXUCOUpl
CykjhzZPqaHKf5hHY5ew6CeY8U/Ll9QMrN8uSPbdYs2yG2saN13WbYgRqqoRMAJS
2OPamiyqm3kOFrXKOVDn5Xc+W35dzdw37yWjPpWuQvSAsgml3jVaSzO0qTuekoDx
rvadnLnxMryD0WQpU0oqV3KJJp3oJud03tGegIJ1FYt8CV0CH+CPUMfgiv4jeen/
qa4yREdjeU00EBlI+6rUL94hjwjxhOPV4PZzgJtiROG2djorQvyANOMCnG9CFFUL
UPvN9r6I1esJiWcNlDr4Al5GcVJ6/O5HKUIw0f8c5Y1EkGsxqT4GVbehKF6CkT11
Wozeel7zWBGeUsVZ+NMcR/xBmBci5fkqN4egVpxafRh2KtX6LdAfEgar6Pe0JfCz
QNMGlpowjoUkzFwRsBicTRc/MRVxH/ubweepeMYmemZLrKWCClGC53tEMO01dJHW
AdrowkPnzTg0G2OWz/nq31DKGYf2bSJc0D1cGGHynqKjX6hzJtjD1jaMsI83pf3x
y/UfNtTa6aGZFy7FS6mf3O4=
-----END CERTIFICATE-----

tailclient.py

#!/usr/bin/env python

## {{{ http://code.activestate.com/recipes/502204/ (r1)
'''Module for running simple proxies.

This module provides a single class that can build
proxy objects capable of being started and stopped.'''

__version__ = '1.0'

import select as _select
import socket as _socket
import ssl as _ssl
import sys as _sys
import thread as _thread
import time
import redis
import sys
import signal
import os.path
import json
from ConfigParser import RawConfigParser
################################################################################

class Proxy:

    'Proxy(bind, connect) -> Proxy'

    FAMILY = _socket.AF_INET
    TYPE = _socket.SOCK_STREAM
    BUFFERSIZE = 2 ** 12

    def __init__(self, bind, connect, keyfile, certfile):
        'Initialize the Proxy object.'
        self.__bind = bind
        self._keyfile = keyfile
        self._certfile = certfile
        self.__connect = connect
        self.__status = False
        self.__thread = False
        self.__lock = _thread.allocate_lock()

    def start(self):
        'Start the Proxy object.'
        self.__lock.acquire()
        self.__status = True
        if not self.__thread:
            self.__thread = True
            _thread.start_new_thread(self.__proxy, ())
        self.__lock.release()

    def stop(self):
        'Stop the Proxy object.'
        self.__lock.acquire()
        self.__status = False
        self.__lock.release()

    def __proxy(self):
        'Private class method.'
        proxy = _socket.socket(self.FAMILY, self.TYPE)
        proxy.bind(self.__bind)
        proxy.listen(5)
        while True:
            client = proxy.accept()[0]
            self.__lock.acquire()
            if not self.__status:
                proxy.close()
                self.__thread = False
                self.__lock.release()
                break
            self.__lock.release()
            server = _socket.socket(self.FAMILY, self.TYPE)
            server = _ssl.wrap_socket(server, 
                                      ca_certs=os.path.expanduser('~/.tailclient/logstash.pem'), 
                                      cert_reqs=_ssl.CERT_REQUIRED, 
                                      certfile=self._certfile, 
                                      keyfile=self._keyfile, 
                                      ssl_version=_ssl.PROTOCOL_SSLv3)
            try:
                server.connect(self.__connect)
            except _ssl.SSLError:
                print("Not authorized")
                sys.exit(1)
            _thread.start_new_thread(self.__serve, (client, server))

    def __serve(self, client, server):
        'Private class method.'
        pairs = {client: server, server: client}
        while pairs:
            read, write, error = _select.select(pairs.keys(), [], [])
            for socket in read:
                string = socket.recv(self.BUFFERSIZE)
                if string:
                    pairs[socket].sendall(string)
                else:
                    pairs[socket].shutdown(_socket.SHUT_WR)
                    socket.shutdown(_socket.SHUT_RD)
                    del pairs[socket]
        client.close()
        server.close()

################################################################################
class TailClient(object):
    """
    Example configuration file:

    [authentication]
    redis_host = logging.my.domain.com 
    access_key_id = XXXXXXXX
    secret_access_key = XXXXXXXX
"""

    def __init__(self, redis_host=None, keyfile=None, certfile=None):
        _redis_host = None
        _certfile = None
        _keyfile = None
        if os.path.exists(os.path.expanduser('~/.tailclient/config.ini')):
            parser = RawConfigParser()
            with open(os.path.expanduser('~/.tailclient/config.ini')) as f:
                parser.readfp(f)
            config = dict(parser.items('authentication'))
            _redis_host = config['redis_host'] if 'redis_host' in config else None
            _keyfile = config['keyfile'] if 'keyfile' in config else None
            _certfile = config['certfile'] if 'certfile' in config else None
        _redis_host = redis_host if redis_host is not None else _redis_host

        if _redis_host is None:
            raise Exception('You must specify the log hostname')
        
        if _certfile is None:
            raise Exception('You must specify a certificate')

        if _keyfile is None:
             raise Exception("You must specify a private key") 
    
        self.proxy = Proxy(('localhost', 22222), (_redis_host, 6379), certfile=_certfile, keyfile=_keyfile)
        self.connection_pool = redis.ConnectionPool(host="localhost", port=22222)

    def start(self):
        self.proxy.start()
        time.sleep(3)
        client = redis.Redis(connection_pool=self.connection_pool)
        pubsub = client.pubsub()
        pubsub.subscribe('logstash')
        for message in pubsub.listen():
            data = json.loads(message['data'])
            fields = data['@fields']
            print("{0[timestamp][0]} - {0[logsource][0]}: {0[message][0]}".format(fields))
    
    def stop(self):
        self.proxy.stop()
        self.connection_pool.disconnect()


if __name__ == '__main__':
    client = TailClient()
    def handler(signum, frame):
        print("Stopping tail process")
        client.stop()
        sys.exit(0)
    signal.signal(signal.SIGINT, handler)
    client.start()