Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
NEW UPDATE: https://github.com/pubnub/python/tree/master/python <-- PubNub Access Manager (PAM) Python Full Library for Granting and Revoking Access in Real-Time on the PubNub Real-Time Network.
import pam
## PubNub Access Manager (PAM)
manager = pam.access(
pubkey="pam",
subkey="pam",
seckey="pam"
)
## Grant User Access
print manager.grant(
channel="my_channel",
authkey="gZW5jb2RlZCBmaWx",
read=True,
write=True,
ttl=5 ## Minutes
)
## Grant User **Presence Access**
## WARNING: The PubNub Dev Console Requires Presence Access
print manager.grant(
channel=[ "my_channel", "my_channel-pnpres" ],
authkey="gZW5jb2RlZCBmaWx",
read=True,
write=True,
ttl=5 ## Minutes
)
## Grant CHANNEL Access (to all users)
## Exclude the authkey and you can global grant access to all.
print manager.grant(
channel="my_channel_all",
read=True,
write=True,
ttl=5 ## Minutes
)
## Forever grant access
## You can grant access forever by setting the ttl param to 0.
print manager.grant(
channel="my_channel-pnpres",
authkey="gZW5jb2RlZCBmaWx",
read=True,
write=True,
ttl=0 ## FOREVER
)
## Revoke User Access
print manager.revoke(
channel="some-other-channel",
authkey="gZW5jb2RlZCBmaWx"
)
## Revoke Channel Access
print manager.revoke(
channel="some-other-channel"
)
## Audit Example
print( 'AUDIT', manager.audit(
channel=[ "my_channel", "my_channel-pnpres" ],
authkey="gZW5jb2RlZCBmaWx"
) )
print( 'AUDIT', manager.audit(
channel="my_channel_all"
) )
print( 'AUDIT', manager.audit() )

PubNub Access Manager Python LIB

Init PAM

manager = pam.access( 
    pubkey="pam",
    subkey="pam",
    seckey="pam"
)

Grant User Access

Grant access to user with authkey of gZW5jb2RlZCBmaWx with read and write access for 5 minute ttl.

print manager.grant(
    channel="my_channel",
    authkey="gZW5jb2RlZCBmaWx",
    read=True,
    write=True,
    ttl=5 ## Minutes
)

Grant User Presence Access

Also grant access to the presence channel (required for PubNub Dev Console).

print manager.grant(
    channel=[ "my_channel", "my_channel-pnpres" ],
    authkey="gZW5jb2RlZCBmaWx",
    read=True,
    write=True,
    ttl=5 ## Minutes
)

Grant CHANNEL Access (to all users)

Exclude the authkey and you can global grant access to all.

print manager.grant(
    channel="my_channel_all",
    read=True,
    write=True,
    ttl=5 ## Minutes
)

Forever Grant Access

You can grant access forever by setting the ttl param to 0.

print manager.grant(
    channel="my_channel-pnpres",
    authkey="gZW5jb2RlZCBmaWx",
    read=True,
    write=True,
    ttl=0 ## FOREVER
)

Revoke User Access

Instantly revoke access to a user.

print manager.revoke(
    channel="some-other-channel",
    authkey="gZW5jb2RlZCBmaWx"
)

Revoke Channel Access

You can also revoke Global Access by excluding the authkey param.

print manager.revoke(
    channel="some-other-channel"
)

Audit Access

You can also Audit Access by running audit().

print manager.audit(
    channel="some-other-channel",
    authkey="gZW5jb2RlZCBmaWx" ## Optional Auth Key
)

Or Audit Access for all users on a channel (exclude the authkey param).

print manager.audit(
    channel="some-other-channel"
)

Or Audit Access for all channels (exclude the authkey and channel params).

print manager.audit()

Dev Console Test Link:

WARNING: PubNub Dev Console Requires Grant on Presence Channel too! You can set the presence access by granting on the suffix of -pnpres channel name.

http://www.pubnub.com/console/?channel=my_channel&sub=pam&pub=pam&sec=pam

from base64 import urlsafe_b64encode
from hashlib import sha256
from urllib import quote
from urllib import urlopen
import json
import hmac
import time
class access():
def __init__( self, pubkey, subkey, seckey ):
self.publish_key = pubkey
self.subscribe_key = subkey
self.secret_key = seckey
def audit( self, channel=False, authkey=False ):
"""Audit R/W Access on a Channel."""
return self._auth( {
"channel" : ",".join(channel) if isinstance(channel, (list, set)) else channel,
"auth" : ",".join(authkey) if isinstance(authkey, (list, set)) else authkey
}, action='audit' )
def grant( self, channel=False, authkey=False, read=True, write=True, ttl=5 ):
"""Grant Access on a Channel."""
return self._auth({
"channel" : ",".join(channel) if isinstance(channel, (list, set)) else channel,
"auth" : ",".join(authkey) if isinstance(authkey, (list, set)) else authkey,
"r" : read and 1 or 0,
"w" : write and 1 or 0,
"ttl" : ttl
})
def revoke( self, channel=False, authkey=False, read=False, write=False, ttl=1 ):
"""Revoke Access on a Channel."""
return self._auth({
"channel" : ",".join(channel) if isinstance(channel, (list, set)) else channel,
"auth" : ",".join(authkey) if isinstance(authkey, (list, set)) else authkey,
"r" : read and 1 or 0,
"w" : write and 1 or 0,
"ttl" : ttl
})
def _sign( self, msg ):
"""Calculate a signature by secret key and message."""
return urlsafe_b64encode(hmac.new(
self.secret_key.encode("utf-8"),
msg.encode("utf-8"),
sha256
).digest())
def _auth( self, query, action='grant' ):
"""Issue an authenticated request."""
if 'timestamp' not in query:
query['timestamp'] = int(time.time())
## Channel Grant?
if 'channel' in query and not query['channel']:
del query['channel']
## Auth Grant?
if 'auth' in query and not query['auth']:
del query['auth']
params = "&".join([
x + "=" + quote(
str(query[x]), safe=""
) for x in sorted(query)
])
signature = self._sign("{subkey}\n{pubkey}\n{action}\n{params}".format(
subkey=self.subscribe_key,
pubkey=self.publish_key,
action=action,
params=params
))
url = ("https://pubsub.pubnub.com/v1/auth/{action}/sub-key/".format(
action=action
) + \
self.subscribe_key + "?" +
params + "&signature=" +
quote(signature, safe=""))
try:
return json.loads(urlopen(url).read())
except:
return {
"error" : True,
"status" : 400,
"message" : "Network Failure",
"payload" : {},
"service" : "Access Manager"
}
@stephenlb

This comment has been minimized.

Copy link
Owner Author

stephenlb commented Dec 28, 2013

@stephenlb

This comment has been minimized.

Copy link
Owner Author

stephenlb commented Sep 16, 2014

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.