Skip to content

Instantly share code, notes, and snippets.

@stephenlb
Last active May 25, 2020 02:43
Show Gist options
  • Save stephenlb/8164176 to your computer and use it in GitHub Desktop.
Save stephenlb/8164176 to your computer and use it in GitHub Desktop.
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
Copy link
Author

stephenlb commented Sep 16, 2014

UPDATE LOCATED AT: https://github.com/pubnub/python <----

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment