User platform¶
Real accounts: sign-up with hashed passwords, sign-in, profile settings, roles and an admin area.
user-platform/
├── app.py # composition root
├── settings.py # DATABASE_URL, session lifetime
├── models/ # database.py, user.py, forms.py
├── services/ # security.py, auth.py, accounts.py, admin.py
└── views/
├── components/ # shell, account_forms
├── pages/ # account.py, admin.py
└── styles/
1. Passwords¶
services/security.py¶
"""Password hashing (standard library scrypt) and safe redirects."""
import hashlib
import hmac
import secrets
_SCRYPT = {"n": 2**14, "r": 8, "p": 1}
def hash_password(password: str) -> str:
"""``scrypt$<salt>$<digest>``, with a random 16-byte salt per password."""
salt = secrets.token_bytes(16)
digest = hashlib.scrypt(password.encode(), salt=salt, **_SCRYPT)
return f"scrypt${salt.hex()}${digest.hex()}"
def verify_password(password: str, stored: str) -> bool:
try:
_, salt, digest = stored.split("$")
except ValueError:
return False
candidate = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), **_SCRYPT)
return hmac.compare_digest(candidate.hex(), digest) # constant time
def safe_next(value: str | None) -> str:
"""Only same-site paths: never redirect to another origin (open-redirect protection)."""
return value if value and value.startswith("/") and not value.startswith("//") else "/"
hashlib.scrypt from the standard library, a random salt per password, and a constant-time
comparison. safe_next refuses //evil.example-style redirects.
2. Who is signed in?¶
services/auth.py¶
"""Who is making this request? Registered with ``app.authenticate(current_user)``."""
from rational import Context
from models import User, UserOut
async def current_user(ctx: Context) -> UserOut | None:
"""The session stores only the user id; the user is loaded fresh on every request, so role
changes and deleted accounts take effect immediately."""
session = ctx.peek_session()
user_id = session.user if session is not None else None
if user_id is None:
return None
async with ctx.db.session() as db:
user = await db.get(User, user_id)
return UserOut.model_validate(user) if user is not None else None
The session only stores the user id. The authenticator loads the user on every request, so a
role change or a deleted account takes effect immediately. app.authenticate(current_user)
installs it, and self.context.user / self.user return its result everywhere.
3. Business rules as field errors¶
async def register(self, name: str, email: str, password: str) -> UserOut:
email = email.strip().lower()
async with self.db.transaction() as db:
if await db.scalar(select(User.id).where(User.email == email)) is not None:
raise FieldError("email", "An account with this email already exists")
first_user = await db.scalar(select(func.count(User.id))) == 0
user = User(
name=name.strip(),
email=email,
password_hash=hash_password(password),
role="admin" if first_user else "member",
)
db.add(user)
await db.flush()
return UserOut.model_validate(user)
FieldError("email", ...) shows the message under the email field, just like a validation
error. Sign-in uses one message for “unknown email” and “wrong password”, so it doesn’t reveal
who has an account.
4. Access control in two places¶
def require_admin(self) -> UserOut:
user = self.require_user()
if not user.is_admin:
raise Forbidden("Administrators only")
return user
views/pages/admin.py¶
"""User administration, under /admin. Members get a 403 page."""
from rational import Avatar, Badge, Button, Card, Column, Heading, Page, Router, Row, Select, State, Text, h, js
from models import ROLES, UserOut
from services import AdminService
from views.components import Shell
from views.styles import CELL, SCROLL_X, TABLE
router = Router(prefix="/admin")
@router.page("/", auth="required")
class Admin(Page):
title = "Admin · User platform"
users = State(factory=list)
async def mount(self):
self.use(AdminService).require_admin() # raises Forbidden -> 403 page for members
await self.reload()
async def reload(self):
self.users = await self.use(AdminService).list_users()
async def set_role(self, user_id: int, role: str):
await self.use(AdminService).set_role(user_id, role)
await self.reload()
async def delete(self, user_id: int):
await self.use(AdminService).delete_user(user_id)
await self.reload()
def row(self, user: UserOut, me: UserOut):
own = user.id == me.id
if own:
role, delete = Badge("you", tone="primary"), None
else:
role = Select(
f"Role of {user.name}",
list(ROLES),
value=user.role,
hide_label=True,
on_change=lambda role, uid=user.id: self.set_role(uid, role),
)
delete = Button(
"Delete",
variant="danger",
size="sm",
aria_label=f"Delete {user.name}",
on_click=[js.confirm(f"Delete {user.name}'s account?"), lambda uid=user.id: self.delete(uid)],
)
person = Row(
Avatar(user.name, size=28),
Column(Text(user.name, weight=600), Text(user.email, size="sm", muted=True), gap=0),
gap="sm",
)
return h("tr", h("td", person, style=CELL), h("td", role, style=CELL), h("td", delete, style=CELL), key=user.id)
def render(self):
me: UserOut = self.context.user
count = len(self.users)
table = h("table", h("tbody", *[self.row(user, me) for user in self.users]), style=TABLE)
return Shell(
Heading("Users", level=1),
Text(f"{count} account{'s' if count != 1 else ''}", muted=True),
Card(h("div", table, style=SCROLL_X)),
)
The page calls require_admin in mount(), so members get a 403 page, and every service
method checks again, so the API is protected too. The router has a prefix: @router.page("/")
is served at /admin.
5. Composition¶
app.py¶
"""User platform: the composition root.
Run: pip install "rational[db]"
python app.py --dev # hot reload, readable errors
python app.py --prod # production server
"""
from rational import App
from models import create_tables, database
from services import AccountService, AdminService, current_user
from views.pages import router
from views.styles import THEME
app = App(
title="User platform",
description="Accounts, sessions and roles, built with Rational Web.",
theme=THEME,
database=database,
login_path="/login",
)
app.authenticate(current_user) # who is signed in, loaded from the database on every request
app.include(router)
app.service(AccountService) # JSON API under /api/account/..., docs at /api/docs
app.service(AdminService) # /api/admin/...
app.on_startup(create_tables)
if __name__ == "__main__":
app.run()