diff options
Diffstat (limited to 'python-twisted')
| -rw-r--r-- | python-twisted/Makefile | 12 | ||||
| -rw-r--r-- | python-twisted/Pubnub.py | 657 | ||||
| -rw-r--r-- | python-twisted/tests/unit-tests.py | 107 | ||||
| -rw-r--r-- | python-twisted/unassembled/Platform.py | 83 |
4 files changed, 853 insertions, 6 deletions
diff --git a/python-twisted/Makefile b/python-twisted/Makefile index 13a5e5a..bf23137 100644 --- a/python-twisted/Makefile +++ b/python-twisted/Makefile @@ -6,13 +6,23 @@ all: build .PHONY: build build: + cat ../common/LICENSE_HEADER > ./Pubnub.py + echo "\n" >> ./Pubnub.py + cat ../common/PubnubCrypto.py >> ./Pubnub.py + echo "\n" >> ./Pubnub.py + cat ../common/PubnubBase.py >> ./Pubnub.py + echo "\n" >> ./Pubnub.py + cat ../common/PubnubCoreAsync.py >> ./Pubnub.py + echo "\n" >> ./Pubnub.py + cat ./unassembled/Platform.py >> ./Pubnub.py find -name "Pubnub*py" | xargs sed -i "s/PubNub\ [0-9]\.[0-9]\.[0-9]/PubNub\ $(VERSION)/g" .PHONY: clean clean: + rm -f Pubnub.py* .PHONY: test test: - python ../common/unit-test-async.py + python tests/unit-tests.py diff --git a/python-twisted/Pubnub.py b/python-twisted/Pubnub.py index 8b29236..9a89aee 100644 --- a/python-twisted/Pubnub.py +++ b/python-twisted/Pubnub.py @@ -8,13 +8,485 @@ ## ----------------------------------- ## PubNub 3.3.4 Real-time Push Cloud API ## ----------------------------------- -import sys -import json + + +from Crypto.Cipher import AES +from Crypto.Hash import MD5 +from base64 import encodestring, decodestring +import hashlib +import hmac + +class PubnubCrypto() : + """ + #** + #* PubnubCrypto + #* + #** + + ## Initiate Class + pc = PubnubCrypto + + """ + + def pad( self, msg, block_size=16 ): + """ + #** + #* pad + #* + #* pad the text to be encrypted + #* appends a padding character to the end of the String + #* until the string has block_size length + #* @return msg with padding. + #** + """ + padding = block_size - (len(msg) % block_size) + return msg + chr(padding)*padding + + def depad( self, msg ): + """ + #** + #* depad + #* + #* depad the decryptet message" + #* @return msg without padding. + #** + """ + return msg[0:-ord(msg[-1])] + + def getSecret( self, key ): + """ + #** + #* getSecret + #* + #* hases the key to MD5 + #* @return key in MD5 format + #** + """ + return hashlib.sha256(key).hexdigest() + + def encrypt( self, key, msg ): + """ + #** + #* encrypt + #* + #* encrypts the message + #* @return message in encrypted format + #** + """ + secret = self.getSecret(key) + Initial16bytes='0123456789012345' + cipher = AES.new(secret[0:32],AES.MODE_CBC,Initial16bytes) + enc = encodestring(cipher.encrypt(self.pad(msg))) + return enc + def decrypt( self, key, msg ): + """ + #** + #* decrypt + #* + #* decrypts the message + #* @return message in decryped format + #** + """ + secret = self.getSecret(key) + Initial16bytes='0123456789012345' + cipher = AES.new(secret[0:32],AES.MODE_CBC,Initial16bytes) + return self.depad((cipher.decrypt(decodestring(msg)))) + + +try: import json +except ImportError: import simplejson as json + import time import hashlib import urllib2 -import uuid -from PubnubCoreAsync import PubnubCoreAsync +import uuid + +class PubnubBase(object): + def __init__( + self, + publish_key, + subscribe_key, + secret_key = False, + cipher_key = False, + ssl_on = False, + origin = 'pubsub.pubnub.com', + UUID = None + ) : + """ + #** + #* Pubnub + #* + #* Init the Pubnub Client API + #* + #* @param string publish_key required key to send messages. + #* @param string subscribe_key required key to receive messages. + #* @param string secret_key optional key to sign messages. + #* @param boolean ssl required for 2048 bit encrypted messages. + #* @param string origin PUBNUB Server Origin. + #* @param string pres_uuid optional identifier for presence (auto-generated if not supplied) + #** + + ## Initiat Class + pubnub = Pubnub( 'PUBLISH-KEY', 'SUBSCRIBE-KEY', 'SECRET-KEY', False ) + + """ + self.origin = origin + self.limit = 1800 + self.publish_key = publish_key + self.subscribe_key = subscribe_key + self.secret_key = secret_key + self.cipher_key = cipher_key + self.ssl = ssl_on + + if self.ssl : + self.origin = 'https://' + self.origin + else : + self.origin = 'http://' + self.origin + + self.uuid = UUID or str(uuid.uuid4()) + + if not isinstance(self.uuid, basestring): + raise AttributeError("pres_uuid must be a string") + + def sign(self, channel, message): + ## Sign Message + if self.secret_key: + signature = hashlib.md5('/'.join([ + self.publish_key, + self.subscribe_key, + self.secret_key, + channel, + message + ])).hexdigest() + else: + signature = '0' + return signature + + def encrypt(self, message): + if self.cipher_key: + pc = PubnubCrypto() + out = [] + if type( message ) == type(list()): + for item in message: + encryptItem = pc.encrypt(self.cipher_key, item ).rstrip() + out.append(encryptItem) + message = json.dumps(out) + elif type( message ) == type(dict()): + outdict = {} + for k, item in message.iteritems(): + encryptItem = pc.encrypt(self.cipher_key, item ).rstrip() + outdict[k] = encryptItem + out.append(outdict) + message = json.dumps(out[0]) + else: + message = json.dumps(pc.encrypt(self.cipher_key, json.dumps(message)).replace('\n','')) + else : + message = json.dumps(message) + + return message; + + def decrypt(self, message): + if self.cipher_key: + pc = PubnubCrypto() + if type( message ) == type(list()): + for item in message: + encryptItem = pc.decrypt(self.cipher_key, item ) + out.append(encryptItem) + message = out + elif type( message ) == type(dict()): + outdict = {} + for k, item in message.iteritems(): + encryptItem = pc.decrypt(self.cipher_key, item ) + outdict[k] = encryptItem + out.append(outdict) + message = out[0] + else: + message = pc.decrypt(self.cipher_key, message) + + return message + + + def publish( self, args ) : + """ + #** + #* Publish + #* + #* Send a message to a channel. + #* + #* @param array args with channel and message. + #* @return array success information. + #** + + ## Publish Example + info = pubnub.publish({ + 'channel' : 'hello_world', + 'message' : { + 'some_text' : 'Hello my World' + } + }) + print(info) + + """ + ## Fail if bad input. + if not (args['channel'] and args['message']) : + return [ 0, 'Missing Channel or Message' ] + + ## Capture User Input + channel = str(args['channel']) + + ## Capture Callback + if args.has_key('callback') : + callback = args['callback'] + else : + callback = None + + #message = json.dumps(args['message'], separators=(',',':')) + message = self.encrypt(args['message']) + + signature = self.sign(channel, message) + + ## Send Message + return self._request({"urlcomponents": [ + 'publish', + self.publish_key, + self.subscribe_key, + signature, + channel, + '0', + message + ]}, callback) + + def presence( self, args ) : + """ + #** + #* presence + #* + #* This is BLOCKING. + #* Listen for presence events on a channel. + #* + #* @param array args with channel and callback. + #* @return false on fail, array on success. + #** + + ## Presence Example + def pres_event(message) : + print(message) + return True + + pubnub.presence({ + 'channel' : 'hello_world', + 'callback' : receive + }) + """ + + ## Fail if missing channel + if not 'channel' in args : + raise Exception('Missing Channel.') + return False + + ## Fail if missing callback + if not 'callback' in args : + raise Exception('Missing Callback.') + return False + + ## Capture User Input + channel = str(args['channel']) + callback = args['callback'] + subscribe_key = args.get('subscribe_key') or self.subscribe_key + + return self.subscribe({'channel': channel+'-pnpres', 'subscribe_key':subscribe_key, 'callback': callback}) + + + def here_now( self, args ) : + """ + #** + #* Here Now + #* + #* Load current occupancy from a channel. + #* + #* @param array args with 'channel'. + #* @return mixed false on fail, array on success. + #* + + ## Presence Example + here_now = pubnub.here_now({ + 'channel' : 'hello_world', + }) + print(here_now['occupancy']) + print(here_now['uuids']) + + """ + channel = str(args['channel']) + + ## Capture Callback + if args.has_key('callback') : + callback = args['callback'] + else : + callback = None + + ## Fail if bad input. + if not channel : + raise Exception('Missing Channel') + return False + + ## Get Presence Here Now + return self._request({"urlcomponents": [ + 'v2','presence', + 'sub_key', self.subscribe_key, + 'channel', channel + ]}, callback); + + + def history( self, args ) : + """ + #** + #* History + #* + #* Load history from a channel. + #* + #* @param array args with 'channel' and 'limit'. + #* @return mixed false on fail, array on success. + #* + + ## History Example + history = pubnub.history({ + 'channel' : 'hello_world', + 'limit' : 1 + }) + print(history) + + """ + ## Capture User Input + limit = args.has_key('limit') and int(args['limit']) or 10 + channel = str(args['channel']) + + ## Fail if bad input. + if not channel : + raise Exception('Missing Channel') + return False + + ## Capture Callback + if args.has_key('callback') : + callback = args['callback'] + else : + callback = None + + ## Get History + return self._request({ "urlcomponents" : [ + 'history', + self.subscribe_key, + channel, + '0', + str(limit) + ] }, callback); + + def detailedHistory(self, args) : + """ + #** + #* Detailed History + #* + #* Load Detailed history from a channel. + #* + #* @param array args with 'channel', optional: 'start', 'end', 'reverse', 'count' + #* @return mixed false on fail, array on success. + #* + + ## History Example + history = pubnub.detailedHistory({ + 'channel' : 'hello_world', + 'count' : 5 + }) + print(history) + + """ + ## Capture User Input + channel = str(args['channel']) + + params = dict() + count = 100 + + if args.has_key('count'): + count = int(args['count']) + + params['count'] = str(count) + + if args.has_key('reverse'): + params['reverse'] = str(args['reverse']).lower() + + if args.has_key('start'): + params['start'] = str(args['start']) + + if args.has_key('end'): + params['end'] = str(args['end']) + + ## Fail if bad input. + if not channel : + raise Exception('Missing Channel') + return False + + ## Capture Callback + if args.has_key('callback') : + callback = args['callback'] + else : + callback = None + + ## Get History + return self._request({ 'urlcomponents' : [ + 'v2', + 'history', + 'sub-key', + self.subscribe_key, + 'channel', + channel, + ],'urlparams' : params }, callback=callback); + + def time(self, args = None) : + """ + #** + #* Time + #* + #* Timestamp from PubNub Cloud. + #* + #* @return int timestamp. + #* + + ## PubNub Server Time Example + timestamp = pubnub.time() + print(timestamp) + + """ + ## Capture Callback + if args and args.has_key('callback') : + callback = args['callback'] + else : + callback = None + time = self._request({'urlcomponents' : [ + 'time', + '0' + ]}, callback) + if time != None: + return time[0] + + + def _encode( self, request ) : + return [ + "".join([ ' ~`!@#$%^&*()+=[]\\{}|;\':",./<>?'.find(ch) > -1 and + hex(ord(ch)).replace( '0x', '%' ).upper() or + ch for ch in list(bit) + ]) for bit in request] + + def getUrl(self,request): + ## Build URL + url = self.origin + '/' + "/".join([ + "".join([ ' ~`!@#$%^&*()+=[]\\{}|;\':",./<>?'.find(ch) > -1 and + hex(ord(ch)).replace( '0x', '%' ).upper() or + ch for ch in list(bit) + ]) for bit in request["urlcomponents"]]) + if (request.has_key("urlparams")): + url = url + '?' + "&".join([ x + "=" + y for x,y in request["urlparams"].iteritems()]) + return url + + try: from hashlib import sha256 digestmod = sha256 @@ -22,6 +494,181 @@ except ImportError: import Crypto.Hash.SHA256 as digestmod sha256 = digestmod.new import hmac + +class PubnubCoreAsync(PubnubBase): + + def start(self): pass + def stop(self): pass + def timeout( self, delay, callback ): + pass + + def __init__( + self, + publish_key, + subscribe_key, + secret_key = False, + cipher_key = False, + ssl_on = False, + origin = 'pubsub.pubnub.com', + uuid = None + ) : + """ + #** + #* Pubnub + #* + #* Init the Pubnub Client API + #* + #* @param string publish_key required key to send messages. + #* @param string subscribe_key required key to receive messages. + #* @param string secret_key required key to sign messages. + #* @param boolean ssl required for 2048 bit encrypted messages. + #* @param string origin PUBNUB Server Origin. + #** + + ## Initiat Class + pubnub = Pubnub( 'PUBLISH-KEY', 'SUBSCRIBE-KEY', 'SECRET-KEY', False ) + + """ + super(PubnubCoreAsync, self).__init__( + publish_key=publish_key, + subscribe_key=subscribe_key, + secret_key=secret_key, + cipher_key=cipher_key, + ssl_on=ssl_on, + origin=origin, + UUID=uuid + ) + + self.subscriptions = {} + self.timetoken = 0 + self.version = '3.4' + self.accept_encoding = 'gzip' + + def subscribe( self, args ) : + """ + #** + #* Subscribe + #* + #* This is NON-BLOCKING. + #* Listen for a message on a channel. + #* + #* @param array args with channel and message. + #* @return false on fail, array on success. + #** + + ## Subscribe Example + def receive(message) : + print(message) + return True + + ## On Connect Callback + def connected() : + pubnub.publish({ + 'channel' : 'hello_world', + 'message' : { 'some_var' : 'text' } + }) + + ## Subscribe + pubnub.subscribe({ + 'channel' : 'hello_world', + 'connect' : connected, + 'callback' : receive + }) + + """ + ## Fail if missing channel + if not 'channel' in args : + return 'Missing Channel.' + + ## Fail if missing callback + if not 'callback' in args : + return 'Missing Callback.' + + ## Capture User Input + channel = str(args['channel']) + callback = args['callback'] + connectcb = args['connect'] + + if 'errorback' in args: + errorback = args['errorback'] + else: + errorback = lambda x: x + + ## New Channel? + if not (channel in self.subscriptions) : + self.subscriptions[channel] = { + 'first' : False, + 'connected' : False, + } + + ## Ensure Single Connection + if self.subscriptions[channel]['connected'] : + return "Already Connected" + + self.subscriptions[channel]['connected'] = 1 + ## SUBSCRIPTION RECURSION + def _subscribe(): + ## STOP CONNECTION? + if not self.subscriptions[channel]['connected']: + return + + def sub_callback(response): + if not self.subscriptions[channel]['first'] : + self.subscriptions[channel]['first'] = True + connectcb() + + ## STOP CONNECTION? + if not self.subscriptions[channel]['connected']: + return + + + + ## PROBLEM? + if not response: + def time_callback(_time): + if not _time: + self.timeout( 1, _subscribe ) + return errorback("Lost Network Connection") + else: + self.timeout( 1, _subscribe) + + ## ENSURE CONNECTED (Call Time Function) + return self.time({ 'callback' : time_callback }) + + self.timetoken = response[1] + _subscribe() + + pc = PubnubCrypto() + out = [] + for message in response[0]: + callback(self.decrypt(message)) + + ## CONNECT TO PUBNUB SUBSCRIBE SERVERS + try: + self._request( { "urlcomponents" : [ + 'subscribe', + self.subscribe_key, + channel, + '0', + str(self.timetoken) + ], "urlparams" : {"uuid":self.uuid} }, sub_callback ) + except : + self.timeout( 1, _subscribe) + return + + ## BEGIN SUBSCRIPTION (LISTEN FOR MESSAGES) + _subscribe() + def unsubscribe( self, args ): + channel = str(args['channel']) + if not (channel in self.subscriptions): + return False + + ## DISCONNECT + self.subscriptions[channel]['connected'] = 0 + self.subscriptions[channel]['timetoken'] = 0 + self.subscriptions[channel]['first'] = False + + from twisted.web.client import getPage from twisted.internet import reactor from twisted.internet.defer import Deferred @@ -29,8 +676,8 @@ from twisted.internet.protocol import Protocol from twisted.web.client import Agent, ContentDecoderAgent, RedirectAgent, GzipDecoder from twisted.web.client import HTTPConnectionPool from twisted.web.http_headers import Headers -from PubnubCrypto import PubnubCrypto from twisted.internet.ssl import ClientContextFactory +from twisted.internet.task import LoopingCall pnconn_pool = HTTPConnectionPool(reactor, persistent=True) pnconn_pool.maxPersistentPerHost = 100 diff --git a/python-twisted/tests/unit-tests.py b/python-twisted/tests/unit-tests.py new file mode 100644 index 0000000..f143a3a --- /dev/null +++ b/python-twisted/tests/unit-tests.py @@ -0,0 +1,107 @@ + +import sys +import time + +sys.path.append('../../common') +sys.path.append('../common') +sys.path.append('..') +sys.path.append('.') + +from PubnubUnitTest import Suite +from Pubnub import Pubnub + +pubnub = Pubnub("demo","demo") + +tests_count = 1 + 2 + 1 +test_suite = Suite(pubnub,tests_count) + +tests = [] + + + +def test_publish(): + channel = "hello" + str(time.time()) + name = "Publish Test" + def success(r): + test_suite.test(r[0] == 1, name) + + def fail(e): + test_suite.test(False, msg , e) + + + pubnub.publish({ + 'channel' : 'hello', + 'message' : 'hi', + 'callback' : success, + 'error' : fail + }) +tests.append(test_publish) + +#""" +def test_subscribe_publish(): + channel = "hello" + str(time.time()) + name = "Subscribe Publish Test" + publish_msg = "This is Pubnub Python-Twisted" + def connect(): + #print 'connect' + def success(r): + test_suite.test(r[0] == 1, name, "publish success") + + def fail(e): + test_suite.test(False, name , "Publish Failed", e) + + pubnub.publish({ + 'channel' : channel, + 'message' : publish_msg, + 'callback' : success, + 'error' : fail + }) + + def callback(r): + test_suite.test(r == publish_msg, name, "message received") + + pubnub.subscribe({ + 'channel' : channel, + 'callback' : callback, + 'connect' : connect + }) +tests.append(test_subscribe_publish) +#""" + +def test_here_now(): + channel = "hello12" #+ str(time.time()) + name = "Here Now Test" + + def connect(): + print 'connect' + def call_here_now(): + print 'call_here_now' + def success(r): + test_suite.test(r['occupancy'] == 1, name, "Here Now success") + + def fail(e): + test_suite.test(False, name , "Here Now Failed", e) + + pubnub.here_now({ + 'channel' : channel, + 'callback' : success, + 'error' : fail + }) + pubnub.timeout(5, call_here_now) + + def callback(r): + pass + print 'Subscribe' + pubnub.subscribe({ + 'channel' : channel, + 'callback' : callback, + 'connect' : connect + }) +tests.append(test_here_now) + + + + +for t in tests: t() + +pubnub.start() diff --git a/python-twisted/unassembled/Platform.py b/python-twisted/unassembled/Platform.py new file mode 100644 index 0000000..e4ae680 --- /dev/null +++ b/python-twisted/unassembled/Platform.py @@ -0,0 +1,83 @@ +from twisted.web.client import getPage +from twisted.internet import reactor +from twisted.internet.defer import Deferred +from twisted.internet.protocol import Protocol +from twisted.web.client import Agent, ContentDecoderAgent, RedirectAgent, GzipDecoder +from twisted.web.client import HTTPConnectionPool +from twisted.web.http_headers import Headers +from twisted.internet.ssl import ClientContextFactory +from twisted.internet.task import LoopingCall + +pnconn_pool = HTTPConnectionPool(reactor, persistent=True) +pnconn_pool.maxPersistentPerHost = 100 +pnconn_pool.cachedConnectionTimeout = 310 + +class Pubnub(PubnubCoreAsync): + + def start(self): reactor.run() + def stop(self): reactor.stop() + def timeout( self, delay, callback ): + reactor.callLater( delay, callback ) + + def __init__( + self, + publish_key, + subscribe_key, + secret_key = False, + cipher_key = False, + ssl_on = False, + origin = 'pubsub.pubnub.com' + ) : + super(Pubnub, self).__init__( + publish_key=publish_key, + subscribe_key=subscribe_key, + secret_key=secret_key, + ssl_on=ssl_on, + origin=origin, + ) + self.headers = {} + self.headers['User-Agent'] = ['Python-Twisted'] + #self.headers['Accept-Encoding'] = [self.accept_encoding] + self.headers['V'] = [self.version] + + def _request( self, request, callback ) : + global pnconn_pool + + ## Build URL + ''' + url = self.origin + '/' + "/".join([ + "".join([ ' ~`!@#$%^&*()+=[]\\{}|;\':",./<>?'.find(ch) > -1 and + hex(ord(ch)).replace( '0x', '%' ).upper() or + ch for ch in list(bit) + ]) for bit in request]) + ''' + url = self.getUrl(request) + agent = ContentDecoderAgent(RedirectAgent(Agent( + reactor, + contextFactory = WebClientContextFactory(), + pool = self.ssl and None or pnconn_pool + )), [('gzip', GzipDecoder)]) + request = agent.request( 'GET', url, Headers(self.headers), None ) + + def received(response): + finished = Deferred() + response.deliverBody(PubNubResponse(finished)) + return finished + + def complete(data): + callback(eval(data)) + + request.addCallback(received) + request.addBoth(complete) + +class WebClientContextFactory(ClientContextFactory): + def getContext(self, hostname, port): + return ClientContextFactory.getContext(self) + +class PubNubResponse(Protocol): + def __init__( self, finished ): + self.finished = finished + + def dataReceived( self, bytes ): + self.finished.callback(bytes) + |