Sessions, Cookies & Flashing

import secrets

from flask import Flask, jsonify, session

app = Flask(__name__)
# Demo only: a key that exists for the lifetime of this process. Real applications
# read a stable key from the environment or from instance configuration.
app.config["SECRET_KEY"] = secrets.token_hex()


@app.get("/books/<int:book_id>")
def show_book(book_id: int):
    recent = session.get("recent", [])
    if book_id in recent:
        recent.remove(book_id)
    recent.insert(0, book_id)
    session["recent"] = recent[:5]  # assigning marks the session as modified
    return jsonify(book=book_id)


@app.get("/recent")
def recent_books():
    return jsonify(recent=session.get("recent", []))


client = app.test_client()
client.get("/books/10")
client.get("/books/20")
client.get("/books/10")
assert client.get("/recent").json == {"recent": [10, 20]}
print(client.get_cookie("session").value[:40], "...")

from flask import Flask, session

demo = Flask(__name__)

# 1. Without a key the session is read-only and empty.
with demo.test_request_context():
    assert dict(session) == {}
    try:
        session["x"] = 1
    except RuntimeError as exc:
        print(exc)

# 2. With a key, in-place mutation of a nested object is not detected.
demo.config["SECRET_KEY"] = secrets.token_hex()
with demo.test_request_context():
    session["cart"] = [1]
    session.modified = False          # pretend the cookie was just saved
    session["cart"].append(2)
    assert session.modified is False  # the change would be lost
    session.modified = True           # tell Flask explicitly ...
    session["cart"] = session["cart"] + [3]  # ... or, better, assign a new object

import base64
import json
import zlib

from flask import Flask, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()


@app.post("/login-demo")
def login_demo():
    session["user_id"] = 42
    session["role"] = "reader"
    return "ok"


client = app.test_client()
client.post("/login-demo")
cookie = client.get_cookie("session").value

payload, timestamp, signature = cookie.rsplit(".", 2)
compressed = payload.startswith(".")
raw = base64.urlsafe_b64decode(payload.lstrip(".") + "=" * (-len(payload) % 4))
if compressed:
    raw = zlib.decompress(raw)
print(json.loads(raw))  # {'role': 'reader', 'user_id': 42}: readable without the key

# Verifying and decoding the intended way needs the key:
serializer = app.session_interface.get_signing_serializer(app)
assert serializer.loads(cookie, max_age=60) == {"user_id": 42, "role": "reader"}

python -c 'import secrets; print(secrets.token_hex())'

import os

from flask import Flask

# In production the shell or the process manager sets this; here the demo does it.
os.environ["FLASK_SECRET_KEY"] = secrets.token_hex()


def create_app() -> Flask:
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_prefixed_env()                       # FLASK_SECRET_KEY -> SECRET_KEY
    app.config.from_pyfile("config.py", silent=True)     # optional, lives in instance/
    if not app.config.get("SECRET_KEY"):
        raise RuntimeError("SECRET_KEY is not configured")
    return app


assert create_app().secret_key

import time
from datetime import timedelta

from flask import Flask, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()
app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(seconds=1)


@app.post("/remember")
def remember():
    session.permanent = True
    session["user_id"] = 5
    return "ok"


@app.get("/who")
def who():
    return {"user_id": session.get("user_id")}


client = app.test_client()
response = client.post("/remember")
set_cookie = response.headers["Set-Cookie"]
assert "Expires=" in set_cookie and "HttpOnly" in set_cookie
assert client.get("/who").json == {"user_id": 5}

time.sleep(2)  # the signature is now older than PERMANENT_SESSION_LIFETIME
assert client.get("/who").json == {"user_id": None}

import secrets
from datetime import timedelta

from flask import Flask, session

app = Flask(__name__)
app.config.update(
    SECRET_KEY=secrets.token_hex(),
    SESSION_COOKIE_NAME="__Host-bookshelf-session",
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE="Lax",
    PERMANENT_SESSION_LIFETIME=timedelta(hours=8),
)


@app.get("/start")
def start():
    session.permanent = True
    session["started"] = True
    return "ok"


header = app.test_client().get("/start", base_url="https://localhost").headers["Set-Cookie"]
print(header)
for attribute in ("__Host-bookshelf-session=", "Secure", "HttpOnly", "SameSite=Lax", "Path=/", "Expires="):
    assert attribute in header, attribute
assert "Domain" not in header

# Partitioned cookies (CHIPS) for an application embedded in third-party iframes.
app.config["SESSION_COOKIE_PARTITIONED"] = True
partitioned = app.test_client().get("/start", base_url="https://localhost").headers["Set-Cookie"]
assert "Partitioned" in partitioned and "Secure" in partitioned

import secrets

from flask import Flask, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()


@app.get("/public")
def public():
    return "same for everybody"


@app.get("/greeting")
def greeting():
    return "hello again" if "name" in session else "hello"


client = app.test_client()
assert "Vary" not in client.get("/public").headers
assert client.get("/greeting").headers["Vary"] == "Cookie"

import secrets
import warnings

from flask import Flask, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()


@app.get("/fill")
def fill():
    session["blob"] = secrets.token_hex(3000)  # random, so it does not compress
    return "ok"


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    app.test_client().get("/fill")
assert any("too large" in str(w.message) for w in caught)
print(str(caught[0].message)[:100])

import secrets

from flask import Flask, abort, request, session
from werkzeug.security import check_password_hash, generate_password_hash

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()

demo_password = secrets.token_urlsafe(12)
USERS = {"alice": generate_password_hash(demo_password)}


@app.post("/login/<name>")
def login(name: str):
    if name not in USERS or not check_password_hash(USERS[name], request.form["password"]):
        abort(401)
    session.clear()            # never keep data from before the login (session fixation)
    session["user"] = name
    return "ok"


@app.post("/logout")
def logout():
    session.clear()
    return "bye"


@app.get("/me")
def me():
    return {"user": session.get("user")}


client = app.test_client()
client.post("/login/alice", data={"password": demo_password})
stolen = client.get_cookie("session").value        # an attacker copied the cookie
client.post("/logout")
assert client.get("/me").json == {"user": None}    # the browser forgot the session ...

attacker = app.test_client()
attacker.set_cookie("session", stolen)
assert attacker.get("/me").json == {"user": "alice"}  # ... but the copy still works

import hashlib
import secrets

from flask import Flask, session
from flask.sessions import SecureCookieSessionInterface


class Sha256SessionInterface(SecureCookieSessionInterface):
    digest_method = staticmethod(hashlib.sha256)


app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()
app.session_interface = Sha256SessionInterface()


@app.get("/set")
def set_value():
    session["theme"] = "dark"
    return "ok"


@app.get("/get")
def get_value():
    return {"theme": session.get("theme")}


client = app.test_client()
client.get("/set")
assert client.get("/get").json == {"theme": "dark"}

import secrets

from flask import Flask, session
from flask.sessions import SessionInterface, SessionMixin
from werkzeug.datastructures import CallbackDict


class ServerSession(CallbackDict, SessionMixin):
    def __init__(self, initial=None, sid=None, new=False):
        def on_update(self):
            self.modified = True

        super().__init__(initial, on_update)
        self.sid = sid
        self.new = new
        self.modified = False


class DictSessionInterface(SessionInterface):
    def __init__(self):
        self.store: dict[str, dict] = {}

    def open_session(self, app, request):
        sid = request.cookies.get(self.get_cookie_name(app))
        if sid and sid in self.store:  # an id we did not issue is never trusted
            return ServerSession(self.store[sid], sid=sid)
        return ServerSession(sid=secrets.token_urlsafe(32), new=True)

    def save_session(self, app, session, response):
        name = self.get_cookie_name(app)
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)
        if session.accessed:
            response.vary.add("Cookie")
        if not session:
            if session.modified:
                self.store.pop(session.sid, None)
                response.delete_cookie(name, domain=domain, path=path)
            return
        if not self.should_set_cookie(app, session):
            return
        self.store[session.sid] = dict(session)
        response.set_cookie(
            name,
            session.sid,
            expires=self.get_expiration_time(app, session),
            httponly=self.get_cookie_httponly(app),
            secure=self.get_cookie_secure(app),
            samesite=self.get_cookie_samesite(app),
            domain=domain,
            path=path,
        )


app = Flask(__name__)
interface = DictSessionInterface()
app.session_interface = interface


@app.post("/cart/<int:book_id>")
def add_to_cart(book_id: int):
    session["cart"] = session.get("cart", []) + [book_id]
    return {"cart": session["cart"]}


@app.post("/cart/clear")
def clear_cart():
    session.clear()
    return "cleared"


client = app.test_client()
client.post("/cart/3")
assert client.post("/cart/4").json == {"cart": [3, 4]}
sid = client.get_cookie("session").value
assert interface.store[sid] == {"cart": [3, 4]}   # the data is on the server ...
assert len(sid) == 43                              # ... the cookie is only an opaque identifier
client.post("/cart/clear")
assert sid not in interface.store

pip install Flask-Session redis

from datetime import timedelta

import fakeredis  # production: redis.Redis.from_url(os.environ["REDIS_URL"])
from flask import Flask, session
from flask_session import Session

redis_client = fakeredis.FakeStrictRedis()

app = Flask(__name__)
app.config.update(
    SESSION_TYPE="redis",
    SESSION_REDIS=redis_client,
    SESSION_KEY_PREFIX="bookshelf:session:",
    PERMANENT_SESSION_LIFETIME=timedelta(hours=8),
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE="Lax",
)
Session(app)


@app.post("/login/<name>")
def login(name: str):
    session["user"] = name                          # (check the password first in real code)
    app.session_interface.regenerate(session)       # fresh identifier: no fixation
    return "ok"


@app.post("/logout")
def logout():
    session.clear()                                 # deletes the record in Redis
    return "bye"


@app.get("/me")
def me():
    return {"user": session.get("user")}


client = app.test_client()
client.post("/login/alice")
sid = client.get_cookie("session").value
key = f"bookshelf:session:{sid}"
assert redis_client.exists(key)
assert 0 < redis_client.ttl(key) <= 8 * 3600       # expiry is enforced by Redis
assert client.get("/me").json == {"user": "alice"}

copy = app.test_client()
copy.set_cookie("session", sid)
client.post("/logout")
assert not redis_client.exists(key)                 # revoked on the server
assert copy.get("/me").json == {"user": None}       # the copied cookie is useless now

import secrets

from flask import Flask, jsonify, make_response, request
from itsdangerous import BadSignature, URLSafeTimedSerializer

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()
signer = URLSafeTimedSerializer(app.config["SECRET_KEY"], salt="bookshelf-page-size")


@app.post("/preferences/theme/<theme>")
def set_theme(theme: str):
    if theme not in {"light", "dark"}:
        return {"error": "unknown theme"}, 400
    response = make_response({"theme": theme})
    response.set_cookie("theme", theme, max_age=60 * 60 * 24 * 365, samesite="Lax", httponly=False)
    return response


@app.post("/preferences/page-size/<int:size>")
def set_page_size(size: int):
    response = make_response({"page_size": size})
    response.set_cookie("page_size", signer.dumps(size), max_age=3600, samesite="Lax", httponly=True)
    return response


@app.get("/preferences")
def preferences():
    theme = request.cookies.get("theme", "light")
    try:
        page_size = signer.loads(request.cookies.get("page_size", ""), max_age=3600)
    except BadSignature:
        page_size = 20          # missing or tampered: fall back to the default
    return jsonify(theme=theme, page_size=page_size)


@app.post("/preferences/reset")
def reset():
    response = make_response("reset")
    response.delete_cookie("theme", samesite="Lax")
    response.delete_cookie("page_size", samesite="Lax")
    return response


client = app.test_client()
client.post("/preferences/theme/dark")
client.post("/preferences/page-size/50")
assert client.get("/preferences").json == {"theme": "dark", "page_size": 50}

client.set_cookie("page_size", "50")             # tampering: an unsigned value
assert client.get("/preferences").json["page_size"] == 20

client.post("/preferences/reset")
assert client.get_cookie("theme") is None
assert client.get("/preferences").json == {"theme": "light", "page_size": 20}

import secrets

from flask import (
    Flask,
    flash,
    get_flashed_messages,
    redirect,
    render_template_string,
    request,
    url_for,
)

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()

LAYOUT = """
{% with messages = get_flashed_messages(with_categories=true) %}
  {% if messages %}
    <ul class="flashes">
      {% for category, message in messages %}
        <li class="{{ category }}">{{ message }}</li>
      {% endfor %}
    </ul>
  {% endif %}
{% endwith %}
<h1>Books</h1>
"""


@app.get("/books")
def book_list():
    return render_template_string(LAYOUT)


@app.post("/books")
def create_book():
    title = request.form.get("title", "").strip()
    if not title:
        flash("A title is required.", "error")
        return redirect(url_for("book_list"))
    flash(f"Saved {title}", "success")
    flash("Remember to add the author.", "warning")
    return redirect(url_for("book_list"))


client = app.test_client()
page = client.post("/books", data={"title": "<b>Dune</b>"}, follow_redirects=True).text
assert 'class="success">Saved &lt;b&gt;Dune&lt;/b&gt;' in page     # autoescaped
assert 'class="warning"' in page

assert "flashes" not in client.get("/books").text                      # shown once only

import secrets

from flask import Flask, flash, get_flashed_messages, redirect, render_template_string

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()


@app.post("/demo")
def demo():
    flash("Disk almost full", "warning")
    flash("Import failed", "error")
    flash("Welcome back")           # the default category is "message"
    return redirect("/show")


@app.get("/show")
def show():
    errors = get_flashed_messages(category_filter=("error",))
    everything = get_flashed_messages(with_categories=True)
    return render_template_string("{{ errors }}|{{ everything }}", errors=errors, everything=everything)


client = app.test_client()
text = client.post("/demo", follow_redirects=True).text
assert "Import failed" in text and "Disk almost full" in text
assert "('message', 'Welcome back')" in text

from flask import message_flashed

recorded = []


def record(sender, message, category, **extra):
    recorded.append((category, message))


with message_flashed.connected_to(record, app):
    client.post("/demo")

assert recorded == [("warning", "Disk almost full"), ("error", "Import failed"), ("message", "Welcome back")]

import secrets

from flask import Flask, flash, redirect, session

app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex()


@app.get("/me")
def me():
    return {"user": session.get("user")}


@app.post("/save")
def save():
    flash("Saved", "success")
    return redirect("/me")


client = app.test_client()

with client.session_transaction() as sess:       # arrange: be logged in without a login view
    sess["user"] = "alice"
assert client.get("/me").json == {"user": "alice"}

client.post("/save")
with client.session_transaction() as sess:       # assert on the raw flashed list
    assert sess["_flashes"] == [("success", "Saved")]

with client:                                     # keep the context to inspect the session after the call
    client.get("/me")
    assert session["user"] == "alice"