Nameko Auth Toy

client.py

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()

requirements.txt

nameko
pyjwt

services.py

"""
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")

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