Security Policy & Authentication

import secrets
from dataclasses import dataclass

from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError

hasher = PasswordHasher()
# A real hash of a random password, used to keep the cost of a failed login
# the same whether or not the user exists.
DUMMY_HASH = hasher.hash(secrets.token_urlsafe(16))


@dataclass
class User:
    id: str
    name: str
    password_hash: str
    is_admin: bool = False


class UserStore:
    """Stands in for a database table; counts how often it is asked."""

    def __init__(self):
        self.by_id = {}
        self.lookups = 0

    def add(self, user_id, name, password, is_admin=False):
        self.by_id[user_id] = User(user_id, name, hasher.hash(password), is_admin)

    def get(self, user_id):
        self.lookups += 1
        return self.by_id.get(user_id)

    def by_name(self, name):
        return next((u for u in self.by_id.values() if u.name == name), None)


def check_password(user, password):
    """Verify against the real hash, or the dummy one for unknown users."""
    stored = user.password_hash if user else DUMMY_HASH
    try:
        hasher.verify(stored, password)
    except (VerificationError, InvalidHashError):
        return False
    return user is not None

from pyramid.authentication import AuthTktCookieHelper
from pyramid.request import RequestLocalCache
from pyramid.security import Allowed, Denied


class ShelvesSecurityPolicy:
    def __init__(self, secret, store, secure=True):
        self.helper = AuthTktCookieHelper(
            secret,
            cookie_name="shelves_auth",
            secure=secure,          # send only over HTTPS
            http_only=True,         # hidden from JavaScript
            samesite="Lax",         # the default, stated for clarity
            timeout=8 * 3600,       # tickets older than 8 h are refused
            reissue_time=900,       # sliding window: re-issue after 15 min
        )
        self.store = store
        self.identity_cache = RequestLocalCache(self.load_identity)

    def load_identity(self, request):
        ticket = self.helper.identify(request)
        if ticket is None:
            return None
        # Re-check the database: a deleted user must stop working at once.
        return self.store.get(ticket["userid"])

    def identity(self, request):
        return self.identity_cache.get_or_create(request)

    def authenticated_userid(self, request):
        user = self.identity(request)
        return None if user is None else user.id

    def permits(self, request, context, permission):
        user = self.identity(request)
        if user is None:
            return Denied("anonymous users have no permissions")
        # The next page replaces this rule with ACLs.
        return Allowed(f"{user.name} is signed in")

    def remember(self, request, userid, **kw):
        return self.helper.remember(request, userid, **kw)

    def forget(self, request, **kw):
        return self.helper.forget(request)

from urllib.parse import quote, urlsplit

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPForbidden, HTTPSeeOther
from pyramid.security import forget, remember
from pyramid.settings import asbool


def safe_next(request, default="/"):
    """Accept only local paths for the post-login redirect."""
    target = request.params.get("next", default)
    parts = urlsplit(target)
    if parts.scheme or parts.netloc or not target.startswith("/") \
            or target.startswith("//") or "\\" in target:
        return default
    return target


def login(request):
    username = request.POST.get("username", "")
    password = request.POST.get("password", "")
    user = request.user_store.by_name(username)
    if not check_password(user, password):
        request.response.status = 401
        return {"error": "invalid credentials"}
    if hasher.check_needs_rehash(user.password_hash):
        user.password_hash = hasher.hash(password)
    return HTTPSeeOther(location=safe_next(request), headers=remember(request, user.id))


def logout(request):
    return HTTPSeeOther(location="/", headers=forget(request))


def me(request):
    user = request.identity
    return {"id": user.id, "name": user.name}


def shelf_edit(request):
    return {"shelf": request.matchdict["owner"], "editing": True}


def forbidden(request):
    if request.is_authenticated:
        request.response.status = 403
        return {"error": "forbidden"}
    # Anonymous: send the visitor to the login page and come back afterwards.
    return HTTPSeeOther(location="/login?next=" + quote(request.path_qs))


def make_app(settings, store):
    with Configurator(settings=settings) as config:
        secure = asbool(settings.get("auth.secure", "true"))
        config.set_security_policy(
            ShelvesSecurityPolicy(settings["auth.secret"], store, secure=secure)
        )
        config.add_request_method(lambda request: store, "user_store", reify=True)
        config.add_route("login", "/login")
        config.add_route("logout", "/logout")
        config.add_route("me", "/me")
        config.add_route("shelf_edit", "/shelves/{owner}/edit")
        config.add_view(login, route_name="login", request_method="POST", renderer="json")
        config.add_view(logout, route_name="logout", request_method="POST")
        config.add_view(me, route_name="me", renderer="json", is_authenticated=True)
        config.add_view(shelf_edit, route_name="shelf_edit", permission="edit", renderer="json")
        config.add_forbidden_view(forbidden, renderer="json")
        return config.make_wsgi_app()

from webtest import TestApp

store = UserStore()
ada_password = secrets.token_urlsafe(12)
store.add("ada", "ada", ada_password)

settings = {"auth.secret": secrets.token_hex(32), "auth.secure": "false"}
app = TestApp(make_app(settings, store))

# Anonymous: the Forbidden View sends the visitor to the login page.
r = app.get("/shelves/ada/edit", status=303)
print(r.status, r.location)

# Wrong password: 401, no cookie.
r = app.post("/login", {"username": "ada", "password": "wrong"}, status=401)
print(r.status, r.json, "Set-Cookie" in r.headers)

# Sign in.
r = app.post("/login", {"username": "ada", "password": ada_password,
                        "next": "/shelves/ada/edit"}, status=303)
print(r.status, r.location)
cookie = r.headers["Set-Cookie"].split(";")[0].split("=")[0]
print("cookie:", cookie, "| attributes:", r.headers["Set-Cookie"].split("; ")[1:])

# The TestApp keeps the cookie jar, so we are now signed in.
print(app.get("/me").json)
print(app.get("/shelves/ada/edit").json)

# Sign out.
app.post("/logout", status=303)
app.get("/me", status=404)   # is_authenticated=True no longer matches
print("lookups so far:", store.lookups)

# The redirect target is checked: only local paths survive.
from pyramid.testing import DummyRequest
for target in ("/shelves/ada", "https://evil.example/", "//evil.example/",
               "/\\evil.example", "javascript:alert(1)"):
    print(repr(target), "->", safe_next(DummyRequest(params={"next": target})))

from pyramid.authentication import SessionAuthenticationHelper
from pyramid.session import SignedCookieSessionFactory


class SessionPolicy:
    def __init__(self, store):
        self.helper = SessionAuthenticationHelper()
        self.store = store
        self.identity_cache = RequestLocalCache(self.load_identity)

    def load_identity(self, request):
        userid = self.helper.authenticated_userid(request)
        return None if userid is None else self.store.get(userid)

    def identity(self, request):
        return self.identity_cache.get_or_create(request)

    def authenticated_userid(self, request):
        user = self.identity(request)
        return None if user is None else user.id

    def permits(self, request, context, permission):
        if self.identity(request) is None:
            return Denied("anonymous")
        return Allowed("signed in")

    def remember(self, request, userid, **kw):
        return self.helper.remember(request, userid, **kw)

    def forget(self, request, **kw):
        return self.helper.forget(request, **kw)


def session_login(request):
    user = request.user_store.by_name(request.POST.get("username", ""))
    if not check_password(user, request.POST.get("password", "")):
        request.response.status = 401
        return {"error": "invalid credentials"}
    request.session.invalidate()          # new session: no session fixation
    headers = remember(request, user.id)  # writes auth.userid into it
    return HTTPSeeOther(location="/", headers=headers)


def session_logout(request):
    request.session.invalidate()
    return HTTPSeeOther(location="/")


def session_me(request):
    return {"user": request.authenticated_userid, "session": dict(request.session)}


def make_session_app(store):
    with Configurator() as config:
        config.set_security_policy(SessionPolicy(store))
        config.set_session_factory(
            SignedCookieSessionFactory(secrets.token_hex(32), httponly=True, samesite="Lax")
        )
        config.add_request_method(lambda request: store, "user_store", reify=True)
        config.add_route("login", "/login")
        config.add_route("logout", "/logout")
        config.add_route("me", "/me")
        config.add_view(session_login, route_name="login", request_method="POST", renderer="json")
        config.add_view(session_logout, route_name="logout", request_method="POST")
        config.add_view(session_me, route_name="me", renderer="json")
        return config.make_wsgi_app()


s_app = TestApp(make_session_app(store))
s_app.post("/login", {"username": "ada", "password": ada_password}, status=303)
print(s_app.get("/me").json)
s_app.post("/logout", status=303)
print(s_app.get("/me").json)

class NoCachePolicy(ShelvesSecurityPolicy):
    def identity(self, request):          # bypasses the cache
        return self.load_identity(request)


class ReifiedPolicy(ShelvesSecurityPolicy):
    def identity(self, request):
        return request.user               # the reified request method


def touch_identity_four_times(request):
    request.identity, request.identity, request.is_authenticated, request.has_permission("edit")
    return {"ok": True}


def count_lookups(policy_factory, reify_user=False):
    counting = UserStore()
    counting.add("ada", "ada", ada_password)
    with Configurator() as config:
        policy = policy_factory(secrets.token_hex(32), counting, secure=False)
        config.set_security_policy(policy)
        if reify_user:
            config.add_request_method(policy.load_identity, "user", reify=True)
        config.add_route("t", "/t")
        config.add_route("login", "/login")
        config.add_view(touch_identity_four_times, route_name="t", renderer="json")
        config.add_view(
            lambda request: HTTPSeeOther("/", headers=remember(request, "ada")),
            route_name="login",
        )
        client = TestApp(config.make_wsgi_app())
    client.get("/login", status=303)
    counting.lookups = 0
    client.get("/t")
    return counting.lookups


print("no cache      :", count_lookups(NoCachePolicy))
print("RequestLocal  :", count_lookups(ShelvesSecurityPolicy))
print("reified method:", count_lookups(ReifiedPolicy, reify_user=True))

import bcrypt
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

pw = secrets.token_urlsafe(12)

# Argon2id: the parameters travel inside the encoded hash.
ph = PasswordHasher()
encoded = ph.hash(pw)
print(encoded.split("$")[1], encoded.count("$"))
print(ph.verify(encoded, pw))
try:
    ph.verify(encoded, pw + "x")
except VerifyMismatchError:
    print("mismatch detected")

# check_needs_rehash is True after you raise the cost: upgrade on the next login.
stronger = PasswordHasher(time_cost=ph.time_cost + 1)
print("rehash needed:", stronger.check_needs_rehash(encoded))

# bcrypt: bytes in, bytes out; 72-byte limit.
digest = bcrypt.hashpw(pw.encode(), bcrypt.gensalt())
print(bcrypt.checkpw(pw.encode(), digest))
try:
    bcrypt.hashpw(b"a" * 80, bcrypt.gensalt())
except ValueError as exc:
    print("bcrypt:", exc)

import hashlib

from pyramid.authentication import extract_http_basic_credentials
from pyramid.httpexceptions import HTTPUnauthorized


def token_digest(token):
    return hashlib.sha256(token.encode()).hexdigest()


class ApiPolicy:
    def __init__(self, store, tokens):
        self.store = store
        self.tokens = tokens                       # sha256 hex digest -> user id
        self.identity_cache = RequestLocalCache(self.load_identity)

    def load_identity(self, request):
        creds = extract_http_basic_credentials(request)
        if creds is not None:
            user = self.store.by_name(creds.username)
            return user if check_password(user, creds.password) else None
        scheme, _, token = request.headers.get("Authorization", "").partition(" ")
        if scheme.lower() == "bearer" and token:
            userid = self.tokens.get(token_digest(token.strip()))
            return None if userid is None else self.store.get(userid)
        return None

    def identity(self, request):
        return self.identity_cache.get_or_create(request)

    def authenticated_userid(self, request):
        user = self.identity(request)
        return None if user is None else user.id

    def permits(self, request, context, permission):
        if self.identity(request) is None:
            return Denied("no valid credentials")
        return Allowed("authenticated API client")

    def remember(self, request, userid, **kw):
        return []                                   # stateless: nothing to remember

    def forget(self, request, **kw):
        return []


def api_forbidden(request):
    if request.is_authenticated:
        request.response.status = 403
        return {"error": "forbidden"}
    return HTTPUnauthorized(
        json_body={"error": "authentication required"},
        headers={"WWW-Authenticate": 'Bearer realm="shelves", charset="UTF-8"'},
    )


def api_shelves(request):
    return {"owner": request.authenticated_userid, "shelves": ["to-read", "favourites"]}


issued_token = secrets.token_urlsafe(32)          # shown to the user once
api_policy = ApiPolicy(store, {token_digest(issued_token): "ada"})

with Configurator() as config:
    config.set_security_policy(api_policy)
    config.add_route("api_shelves", "/api/shelves")
    config.add_view(api_shelves, route_name="api_shelves", permission="read", renderer="json")
    config.add_forbidden_view(api_forbidden, renderer="json")
    api = TestApp(config.make_wsgi_app())

r = api.get("/api/shelves", status=401)
print(r.status, r.headers["WWW-Authenticate"])
print(api.get("/api/shelves", headers={"Authorization": f"Bearer {issued_token}"}).json)
print(api.get("/api/shelves", headers={"Authorization": "Bearer " + secrets.token_urlsafe(32)},
              status=401).status)

from base64 import b64encode
basic = b64encode(f"ada:{ada_password}".encode()).decode()
print(api.get("/api/shelves", headers={"Authorization": f"Basic {basic}"}).json)

import os


def load_secrets(environ=os.environ, production=True):
    names = ("SHELVES_AUTH_SECRET", "SHELVES_SESSION_SECRET")
    values = {}
    for name in names:
        value = environ.get(name)
        if not value:
            if production:
                raise RuntimeError(f"{name} must be set in production")
            value = secrets.token_hex(32)       # development only: per process
        if len(value) < 32:
            raise RuntimeError(f"{name} is too short; use at least 32 random characters")
        values[name] = value
    if len(set(values.values())) != len(values):
        raise RuntimeError("each subsystem needs its own secret; do not share them")
    return values


# Development: random secrets, distinct by construction.
dev = load_secrets({}, production=False)
print(sorted(dev), len(set(dev.values())))

# Production: a missing or shared secret stops the application at start-up.
for env in ({}, {"SHELVES_AUTH_SECRET": "x" * 40, "SHELVES_SESSION_SECRET": "x" * 40}):
    try:
        load_secrets(env)
    except RuntimeError as exc:
        print("refused:", exc)

import warnings

from pyramid.authentication import AuthTktAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy


def groupfinder(userid, request):
    # the "callback": None for unknown users, otherwise a list of principals
    return ["group:editors"] if userid == "ada" else None


with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    with Configurator() as legacy:
        legacy.set_authentication_policy(
            AuthTktAuthenticationPolicy(secrets.token_hex(32), callback=groupfinder)
        )
        legacy.set_authorization_policy(ACLAuthorizationPolicy())
        legacy.add_route("x", "/x")
        legacy.add_view(lambda request: {"user": request.authenticated_userid},
                        route_name="x", renderer="json")
        legacy_app = TestApp(legacy.make_wsgi_app())
        legacy_app.get("/x")
print(sorted({w.category.__name__ for w in caught}))
print(sorted({str(w.message).split(".")[0][:70] for w in caught if "deprecated" in str(w.message)}))

class ShimSecurityPolicy:
    def __init__(self, authn_policy, authz_policy):
        self.authn_policy = authn_policy
        self.authz_policy = authz_policy

    def identity(self, request):
        return self.authn_policy.authenticated_userid(request)

    def authenticated_userid(self, request):
        return self.authn_policy.authenticated_userid(request)

    def permits(self, request, context, permission):
        principals = self.authn_policy.effective_principals(request)
        return self.authz_policy.permits(context, principals, permission)

    def remember(self, request, userid, **kw):
        return self.authn_policy.remember(request, userid, **kw)

    def forget(self, request, **kw):
        return self.authn_policy.forget(request)


with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    shim = ShimSecurityPolicy(
        AuthTktAuthenticationPolicy(secrets.token_hex(32), callback=groupfinder),
        ACLAuthorizationPolicy(),
    )
    with Configurator() as config:
        config.set_security_policy(shim)
        config.add_route("x", "/x")
        config.add_view(lambda request: {"ok": True}, route_name="x", renderer="json")
        TestApp(config.make_wsgi_app()).get("/x")
print("shim app built and served a request")