Skip to content

Instantly share code, notes, and snippets.

@mattbennett
Created April 14, 2017 15:00
Show Gist options
  • Save mattbennett/8a67956ea78d4481d5ea9072e770ba4b to your computer and use it in GitHub Desktop.
Save mattbennett/8a67956ea78d4481d5ea9072e770ba4b to your computer and use it in GitHub Desktop.
Nameko Auth Toy
from nameko.standalone.rpc import ClusterRpcProxy
config = {
'AMQP_URI': 'amqp://guest:guest@localhost:5672/'
}
with ClusterRpcProxy(config) as rpc:
session_token = rpc.auth.login("admin", "secret")
with ClusterRpcProxy(config, context_data={'session': session_token}) as rpc:
rpc.a.foo()
"""
Toy authentication and authorization example using JSON Web Tokens
"""
from nameko.extensions import DependencyProvider
from nameko.rpc import rpc, RpcProxy
import jwt
JWT_SECRET = "secret"
class NotAuthenticated(Exception):
pass
class AuthService:
name = "auth"
@rpc
def login(self, username, password):
if password == "secret":
# probably add roles or permissions for the user here
perms = []
roles = []
token = jwt.encode({
'username': username,
'permissions': [],
'roles': []
}, JWT_SECRET)
return token
raise NotAuthenticated()
class Authorization(DependencyProvider):
""" DependencyProvider giving services access to the current session.
"""
class Api:
def __init__(self, token):
self._token = token
self._session = None
@property
def session(self):
if self._session is None:
# lazily decode the token
self._session = jwt.decode(self._token, JWT_SECRET)
return self._session
@property
def username(self):
return self.session['username']
def user_is(self, role):
return role in self.session['roles']
def user_can(self, perm):
return perm in self.session['permissions']
def get_dependency(self, worker_ctx):
return Authorization.Api(worker_ctx.context_data.get('session'))
class ServiceA:
name = 'a'
b_rpc = RpcProxy('b')
@rpc
def foo(self):
self.b_rpc.bar()
class ServiceB:
name = 'b'
auth = Authorization()
@rpc
def bar(self):
print("Logged in user: {}".format(self.auth.username))
if self.auth.user_is("admin"):
print("User is an admin")
else:
print("User is not an admin")
@mattbennett
Copy link
Author

Usage:

$ python client.py
$ nameko run services
starting services: a, b, auth
Connected to amqp://guest:**@127.0.0.1:5672//
Connected to amqp://guest:**@127.0.0.1:5672//
Connected to amqp://guest:**@127.0.0.1:5672//
Logged in user: admin
User is not an admin

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