Sessions & Flash Messages

import json
import os
import secrets

from pyramid.config import Configurator
from pyramid.response import Response
from pyramid.session import SignedCookieSessionFactory
from webtest import TestApp

# In production the secret comes from the environment or a secret store, is
# identical in every worker process, and is never committed to the repository.
SESSION_SECRET = os.environ.get('SHELVES_SESSION_SECRET') or secrets.token_hex(32)


def preferences(request):
    session = request.session
    session['visits'] = session.get('visits', 0) + 1
    if 'sort' in request.params:
        session['sort'] = request.params['sort']
    body = {'visits': session['visits'], 'sort': session.get('sort', 'title'), 'new': session.new}
    return Response(json_body=body)


def make_app(views=None, session_factory=None, **factory_options):
    """Build a WSGI app; reused by the other examples on this page."""
    if session_factory is None:
        session_factory = SignedCookieSessionFactory(SESSION_SECRET, **factory_options)
    config = Configurator(session_factory=session_factory)
    for name, view in (views or {'preferences': preferences}).items():
        config.add_route(name, '/' + name)
        config.add_view(view, route_name=name)
    return config.make_wsgi_app()


app = TestApp(make_app())
print(app.get('/preferences?sort=author').json)
print(app.get('/preferences').json)
print(app.get('/preferences').json)

# A different browser (a fresh cookie jar) has its own session.
print(TestApp(make_app()).get('/preferences').json)

# Without a session factory request.session does not exist.
bare = Configurator()
bare.add_route('p', '/p')
bare.add_view(preferences, route_name='p')
try:
    TestApp(bare.make_wsgi_app()).get('/p')
except AttributeError as exc:
    print('no factory:', type(exc).__name__)

import base64
import hashlib
import json

app = TestApp(make_app())
app.get('/preferences?sort=author')
cookie = app.cookies['session']

# value = urlsafe-base64( HMAC-SHA512 signature (64 bytes) + JSON payload )
raw = base64.urlsafe_b64decode(cookie + '=' * (-len(cookie) % 4))
digest_size = hashlib.sha512().digest_size
signature, payload = raw[:digest_size], raw[digest_size:]
last_access, created, data = json.loads(payload)
print('readable by anyone:', data)

# Editing the payload invalidates the signature, so the session is dropped.
forged = payload.replace(b'author', b'AUTHOR')
forged_cookie = base64.urlsafe_b64encode(signature + forged).decode()
attacker = TestApp(make_app())
attacker.set_cookie('session', forged_cookie)
print('forged cookie accepted?', attacker.get('/preferences').json['visits'] != 1)

# A cookie signed with another secret is rejected the same way.
other = TestApp(make_app(session_factory=SignedCookieSessionFactory(secrets.token_hex(32))))
other.set_cookie('session', cookie)
print('foreign cookie accepted?', other.get('/preferences').json['visits'] != 1)

production = SignedCookieSessionFactory(
    SESSION_SECRET,
    cookie_name='__Host-shelves',
    secure=True,
    httponly=True,
    samesite='Lax',
    timeout=600,
    reissue_time=60,
    max_age=None,
    set_on_exception=False,
)
app = TestApp(make_app(session_factory=production), extra_environ={'wsgi.url_scheme': 'https'})
response = app.get('/preferences?sort=author')
header = response.headers['Set-Cookie']
print(header.split('=', 1)[0])
print(sorted(part.strip() for part in header.split(';')[1:]))

from pyramid.httpexceptions import HTTPSeeOther


def anonymous_visit(request):
    request.session['cart_note'] = 'saved before login'
    return Response('ok')


def login(request):
    request.session.invalidate()                 # drop everything stored while anonymous
    request.session['user_id'] = 42              # then record the identity
    return HTTPSeeOther(location=request.route_url('whoami'))


def logout(request):
    request.session.invalidate()
    return HTTPSeeOther(location=request.route_url('whoami'))


def whoami(request):
    return Response(json_body=dict(request.session))


views = {'anonymous_visit': anonymous_visit, 'login': login, 'logout': logout, 'whoami': whoami}
app = TestApp(make_app(views))
app.get('/anonymous_visit')
print('before login:', app.get('/whoami').json)
app.get('/login').follow()
print('after login: ', app.get('/whoami').json)
app.get('/logout').follow()
print('after logout:', app.get('/whoami').json)

from unittest import mock

import pyramid.session


def read_visits(request):
    session = request.session
    session['visits'] = session.get('visits', 0) + 1
    return Response(json_body={'visits': session['visits'], 'new': session.new})


def read_only(request):
    return Response(json_body={'visits': request.session.get('visits', 0)})


views = {'visits': read_visits, 'peek': read_only}


def at(clock, app, path):
    with mock.patch.object(pyramid.session.time, 'time', return_value=clock):
        return app.get(path)


start = 1_800_000_000

# 1. Idle for longer than the timeout: the data is discarded, but new stays False.
app = TestApp(make_app(views, timeout=600))
at(start, app, '/visits')
print('within timeout :', at(start + 300, app, '/visits').json)
print('after timeout  :', at(start + 300 + 601, app, '/visits').json)

# 2. reissue_time=None: reading does not refresh the cookie, so the session ends
#    `timeout` seconds after the last change even though the visitor kept reading.
app = TestApp(make_app(views, timeout=600, reissue_time=None))
at(start, app, '/visits')
at(start + 500, app, '/peek')
print('no read refresh :', at(start + 700, app, '/peek').json)

# 3. The default reissue_time=0: every read slides the window forward.
app = TestApp(make_app(views, timeout=600))
at(start, app, '/visits')
at(start + 500, app, '/peek')
print('sliding window  :', at(start + 700, app, '/peek').json)

# 4. A read-only request with reissue_time=None sends no Set-Cookie header at all.
app = TestApp(make_app(views, reissue_time=None))
at(start, app, '/visits')
print('Set-Cookie on read:', 'Set-Cookie' in at(start + 5, app, '/peek').headers)

from pyramid.testing import DummyRequest

def flash_demo(request):
    session = request.session
    session.flash('Welcome back')                              # default queue
    session.flash('Shelf created', 'success')
    session.flash('Shelf created', 'success', allow_duplicate=False)   # ignored: already queued
    session.flash('Shelf created', 'success')                  # appended: duplicates are allowed by default
    session.flash({'level': 'warning', 'text': 'Quota almost used'}, 'banner')
    result = {
        'peek success': session.peek_flash('success'),
        'peek again': session.peek_flash('success'),
        'pop success': session.pop_flash('success'),
        'pop success again': session.pop_flash('success'),
        'default queue': session.peek_flash(),
        'banner': session.peek_flash('banner'),
    }
    return Response(json_body=result)


app = TestApp(make_app({'flash_demo': flash_demo}))
for key, value in app.get('/flash_demo').json.items():
    print(f'{key:18}', value)

import jinja2
from pyramid.httpexceptions import HTTPSeeOther

env = jinja2.Environment(autoescape=True)
PAGE = env.from_string(
    '{% for level in levels %}'
    '{% for text in messages[level] %}<p class="flash {{ level }}">{{ text }}</p>{% endfor %}'
    '{% endfor %}'
    '<ul>{% for shelf in shelves %}<li>{{ shelf }}</li>{% endfor %}</ul>'
)
SHELVES = ['Fiction']


def list_shelves(request):
    levels = ('success', 'error')
    messages = {level: request.session.pop_flash(level) for level in levels}
    return Response(PAGE.render(levels=levels, messages=messages, shelves=SHELVES), content_type='text/html')


def create_shelf(request):
    name = request.POST.get('name', '').strip()
    if not name:
        request.session.flash('A shelf needs a name', 'error')
    else:
        SHELVES.append(name)
        request.session.flash(f'Shelf {name} created', 'success')
    return HTTPSeeOther(location=request.route_url('list_shelves'))     # always redirect after POST


config = Configurator(session_factory=SignedCookieSessionFactory(SESSION_SECRET, httponly=True))
config.add_route('list_shelves', '/shelves')
config.add_route('create_shelf', '/shelves/new')
config.add_view(list_shelves, route_name='list_shelves', request_method='GET')
config.add_view(create_shelf, route_name='create_shelf', request_method='POST')
app = TestApp(config.make_wsgi_app())

page = app.post('/shelves/new', {'name': 'Poetry'}).follow()
print('after create:', [p for p in page.html.find_all('p')][0].text)
print('refresh     :', [p.text for p in app.get('/shelves').html.find_all('p')])
page = app.post('/shelves/new', {'name': ''}).follow()
print('after error :', [p.attrs['class'] + [p.text] for p in page.html.find_all('p')])
page = app.post('/shelves/new', {'name': '<script>alert(1)</script>'}).follow()
print('escaped     :', '<script>' not in page.text, '&lt;script&gt;' in page.text)

python -m pip install pyramid_nacl_session

from pyramid.session import JSONSerializer
from pyramid_nacl_session import EncryptedCookieSessionFactory

nacl_secret = secrets.token_bytes(32)        # exactly 32 random bytes; share it between workers via config
factory = EncryptedCookieSessionFactory(nacl_secret, serializer=JSONSerializer(), httponly=True)
app = TestApp(make_app(session_factory=factory))
app.get('/preferences?sort=author')
cookie = app.cookies['session']
raw = base64.urlsafe_b64decode(cookie + '=' * (-len(cookie) % 4))
print('author' in raw.decode('latin-1'), len(raw) > 0)         # False: the sort order is not visible
print(app.get('/preferences').json)                          # the application still reads it

import time

from pyramid.interfaces import ISession
from zope.interface import implementer


class MemorySessionStore:
    """Stand-in for Redis or SQL: get, set and delete JSON strings by id."""

    def __init__(self):
        self._data = {}

    def get(self, sid):
        return self._data.get(sid)

    def set(self, sid, value):
        self._data[sid] = value

    def delete(self, sid):
        self._data.pop(sid, None)


def ServerSideSessionFactory(store, cookie_name='sid', timeout=1200, secure=False):
    @implementer(ISession)
    class ServerSideSession(dict):
        def __init__(self, request):
            super().__init__()
            self.request = request
            self.sid = request.cookies.get(cookie_name)
            self.new = True
            self.created = int(time.time())
            self._invalidated = False
            record = store.get(self.sid) if self.sid else None
            if record is not None:
                saved = json.loads(record)
                if time.time() - saved['accessed'] <= timeout:
                    self.update(saved['data'])
                    self.created = saved['created']
                    self.new = False
                else:
                    store.delete(self.sid)
            if self.new:
                self.sid = None
            request.add_response_callback(self._save)   # runs only if the session was used

        # ISession
        def changed(self):
            pass                                         # everything is saved at the end of the request

        def invalidate(self):
            if self.sid:
                store.delete(self.sid)
            self.sid = None
            self._invalidated = True
            self.clear()

        def flash(self, msg, queue='', allow_duplicate=True):
            queue_list = self.setdefault('_f_' + queue, [])
            if allow_duplicate or msg not in queue_list:
                queue_list.append(msg)

        def pop_flash(self, queue=''):
            return self.pop('_f_' + queue, [])

        def peek_flash(self, queue=''):
            return self.get('_f_' + queue, [])

        def new_csrf_token(self):
            token = secrets.token_hex(20)
            self['_csrft_'] = token
            return token

        def get_csrf_token(self):
            return self.get('_csrft_') or self.new_csrf_token()

        # persistence
        def _save(self, request, response):
            if not self and self.sid is None:
                if self._invalidated:
                    response.delete_cookie(cookie_name)
                return                                   # nothing to store, no cookie
            if self.sid is None:
                self.sid = secrets.token_urlsafe(32)
            record = {'accessed': time.time(), 'created': self.created, 'data': dict(self)}
            store.set(self.sid, json.dumps(record))
            response.set_cookie(cookie_name, self.sid, httponly=True, samesite='Lax', secure=secure)

    return ServerSideSession


store = MemorySessionStore()
app = TestApp(make_app(views={'visits': read_visits, 'peek': read_only, 'login': login, 'whoami': whoami},
                       session_factory=ServerSideSessionFactory(store)))
print(app.get('/visits').json, app.get('/visits').json)
sid = app.cookies['sid']
print('cookie holds only an id:', len(sid), 'characters; server record:', json.loads(store.get(sid))['data'])
app.get('/login').follow()
print('id rotated on login:', app.cookies['sid'] != sid, '| old id still stored:', store.get(sid) is not None)

# Revocation: delete the record on the server and the cookie is worthless.
store.delete(app.cookies['sid'])
print('after revocation:', app.get('/whoami').json)

from pyramid.testing import DummyRequest


def create_shelf_unit(request):
    request.session.flash('Shelf created', 'success')
    return {'name': request.POST['name']}


request = DummyRequest(post={'name': 'Poetry'})
print(create_shelf_unit(request), request.session.peek_flash('success'))
print(type(request.session).__name__)

# Functional test of the cookie jar: reset() makes a new visitor.
app = TestApp(make_app())
app.get('/preferences?sort=author')
print(app.get('/preferences').json['visits'])
app.reset()
print(app.get('/preferences').json['visits'])