Our application currently allows anyone with access to the server to view and create blogposts. We’ll change that to allow only registered users to create blogposts.

So we need a service to create new users, and we have to implement access control.

As in Create Blogposts we will:

  • create a database table
  • create a SQL Alchemy model
  • implement a user service
  • restrict the “create blogpost” endpoint to registered users

Table users

We want to create a very basic user table with these columns:

- id: primary key
- name: username
- password: the users password

Open the file <project-dir>/etc/crate_setup.sql and add the following SQL statement:

create table users (
    id string primary key,
    name string,
    password string

Add the following line to the create_cleanup.sql file:

drop table users

Ensure that crate is running and then run crate_setup:

$sh bin/crate_setup


You might get an error message because the blogpost table is already created. This error can be ignored.

SQL Alchemy User Model

Create a new user module similar to the blogpost module we created before:

$sh mkdir user
$sh touch user/

Create the file user/ with the following contents:

from sqlalchemy import Column, String
from microblog.model import Base, genuuid

class User(Base):

    __tablename__ = 'users'

    id = Column(String, default=genuuid, primary_key=True)
    name = Column('name', String, nullable=False)
    password = Column('password', String, nullable=False)

Also move the genuuid function from microblog/blogpost/ to microblog/ and change the import at the top of microblog/blogpost/ to:

from microblog.model import Base, genuuid

Enabling Authentication Policy

By default, Pyramid does not enable any authentication policy. All views are accessible by anonymous users. In order to begin protecting views from execution based on security settings, you need to enable an authentication policy.

To do that we have to change the instantiation of the Configurator in microblog.server.app_factory:

from pyramid.authentication import AuthTktAuthenticationPolicy
def app_factory(global_config, **settings):
    config = Configurator(settings=settings,

Also create an AuthACLFactory in the

from pyramid import security
class AuthACLFactory(object):

    def __acl__(self):
        user = security.authenticated_userid(self.request)
        if user is not None:
            return [(security.Allow, security.Everyone, security.Authenticated)]
        return []

    def __init__(self, request):
        self.request = request

The factory checks if the user is logged in and adds the permissions Allow, Everyone and Authenticated to the current context.

For details about authentication and authorization in pyramid see: Pyramid security.

The user service

Create the file user/ with the following contents:

from pyramid import security
from sqlalchemy.orm.exc import NoResultFound
from import RestService, rpcmethod_route, rpcmethod_view
from hashlib import sha1

from microblog.server import AuthACLFactory
from microblog.user.model import User
from ..model import DBSession, refresher

class UserService(object):

    def __init__(self, request):
        self.request = request

    def list(self):
        """ List all registered users """
        query = DBSession.query(User).order_by(
        users = query.all()
        result = []
        for user in users:
        return {"data": {"users": result}}

    @rpcmethod_route(route_suffix="/register", request_method="POST")
    def register(self, name, password):
        """ Register a new user """
        user = User() = name
        user.password = self.hash_password(password)
        return {"status": "success"}

    @rpcmethod_route(route_suffix="/login", request_method="POST")
    def login(self, name, password):
        """ Login the given user """
        hashed_pwd = self.hash_password(password)
        query = DBSession.query(User).filter( == name,
                                             User.password == hashed_pwd)
        status = 'failed'
            user =
            headers = security.remember(self.request,
            status = 'success'
        except NoResultFound:
            self.request.response.status = 401
            return {"status": status}

    def hash_password(self, password):
        if isinstance(password, unicode):
            password_8bit = password.encode('UTF-8')
            password_8bit = password
        hashed = sha1('salt' + password_8bit).hexdigest()
        if not isinstance(hashed, unicode):
            hashed = hashed.decode('UTF-8')
        return hashed

def includeme(config):
    config.add_route('users', '/users', static=True, factory=AuthACLFactory)

The list method

In the list method in the blogposts service we fetch all users and build a result list which contains the usernames.

Sometimes it’s required to pass arguments to the view. For this case use the decorator rpcmethod_view. Because we don’t want strangers to see the user list we use the rpcmethod_view decorator to pass the required permission to the view.

It’s also possible to restrict access to the whole service:

@RestService('users', permission=security.Authenticated)

We shouldn’t do that, because register and login must be accessible for any user.

The register method

Like we did in the blogposts service we create a new user here and add it to the DBSession. We don’t have to flush the DBSession manually because this time we don’t want to return the users id in the response.

We pass the parameter route_suffix to the rpcmmethod_route decorator so the register uri is /users/register.

The login method

The login method tries to query a user with the given name and the hashed password. If such an user exists security.remember is called and the return header that contains the authentication cookie is added to the response header.

If the user is not found we change the response status code to 401 UNAUTHORIZED and return an error status.


As in the blogpost service we define the route in the includeme function. We also pass the AuthACLFactory we created earlier, to determine the users permission.

Test the app

Scan and include the new modules within the app_factory:


After restarting the app try to request the users list:

$sh curl -XGET localhost:9210/users
  <title>403 Forbidden</title>
  <h1>403 Forbidden</h1>
  Access was denied to this resource.<br/><br/>
Unauthorized: UserService failed permission check


So register a new user:

curl -XPOST localhost:9210/users/register -d '{"name": "lovely", "password": "1234"}' -H "Content-Type: application/json"
{"status": "success"}


curl -XPOST localhost:9210/users/login -d '{"name": "lovely", "password": "1234"}' -H "Content-Type: application/json" -vv
< Set-Cookie: auth_tkt="<authentication_token>"; Path=/; Domain=.localhost
{"status": "success"}

Copy the authentication token and fetch the user list as authenticated user:

curl -XGET localhost:9210/users -H 'Cookie: auth_tkt="<authentication_token>"'
{"data": {"users": [{"name": "lovely"}]}}

Restrict Creating Blogposts

To restrict the create method of the blogpost service just add this decorator:


It’s also necessary to pass the factory when adding the route:

config.add_route('blogposts', '/blogposts', static=True, factory=AuthACLFactory)

Because only authenticated users are allowed to create a new post we can adapt the create method so the correct username gets assigned to creator:

user = security.authenticated_userid(self.request)
blogpost.creator = user

Test the application

Restart the app and create a new post:

$sh curl -XPOST localhost:9210/blogposts -H 'Cookie: auth_tkt="<token>"' -d '{"text": "authenticated post"}' -H "Content-Type: application/json"

$sh curl -XGET localhost:9210/blogposts
    "data": {
        "blogposts": [
                "created": "...",
                "creator": "lovely",
                "id": "...",
                "text": "authenticated post"