feat(web): add shared pagination and CSRF helpers
Generate filter-preserving page links and validate tokens on web forms.
This commit is contained in:
1 parent
3c70f364f8
commit
5ded340997
1 file changed
+67
@@ -0,0 +1,67 @@
|
|||||||
|
"""Small web controls adapted from RexiMemo's paged settings and lists."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from html import escape
|
||||||
|
from http.cookies import SimpleCookie, CookieError
|
||||||
|
from urllib.parse import urlencode
|
||||||
|
import math
|
||||||
|
import re
|
||||||
|
import secrets
|
||||||
|
|
||||||
|
|
||||||
|
def page_number(value, total, per_page):
|
||||||
|
try:
|
||||||
|
page = int(value)
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
page = 1
|
||||||
|
return max(1, min(page, max(1, math.ceil(total / per_page))))
|
||||||
|
|
||||||
|
|
||||||
|
def pagination(page, total, per_page, path, **params):
|
||||||
|
pages = max(1, math.ceil(total / per_page))
|
||||||
|
if pages <= 1:
|
||||||
|
return ""
|
||||||
|
def link(number, label, title):
|
||||||
|
target = path + "?" + urlencode({**params, "page": number})
|
||||||
|
return f'<a href="{escape(target, quote=True)}" aria-label="{title}">{label}</a>'
|
||||||
|
first = '<svg viewBox="0 0 20 16" aria-hidden="true"><path d="m9 3-5 5 5 5m7-10-5 5 5 5"/></svg>'
|
||||||
|
last = '<svg viewBox="0 0 20 16" aria-hidden="true"><path d="m4 3 5 5-5 5m7-10 5 5-5 5"/></svg>'
|
||||||
|
parts = [link(1, first, "First page"), link(page-1, "‹", "Previous page")] if page > 1 else []
|
||||||
|
for number in range(max(1, page-2), min(pages, page+2)+1):
|
||||||
|
parts.append(f'<span aria-current="page">{number}</span>' if number == page else link(number, str(number), f"Page {number}"))
|
||||||
|
if page < pages:
|
||||||
|
parts.extend([link(page+1, "›", "Next page"), link(pages, last, "Last page")])
|
||||||
|
return '<nav class="pagination" aria-label="Pages">' + ''.join(parts) + '</nav>'
|
||||||
|
|
||||||
|
|
||||||
|
def csrf_token(request, create=True):
|
||||||
|
if getattr(request, "_ose_csrf", ""):
|
||||||
|
return request._ose_csrf
|
||||||
|
headers = request.getAllHeaders()
|
||||||
|
cookie = next((value for key, value in headers.items() if str(key.decode() if isinstance(key, bytes) else key).lower() == "cookie"), "")
|
||||||
|
jar = SimpleCookie()
|
||||||
|
try:
|
||||||
|
jar.load(cookie.decode() if isinstance(cookie, bytes) else cookie)
|
||||||
|
token = jar["ose_csrf"].value if "ose_csrf" in jar else ""
|
||||||
|
except CookieError:
|
||||||
|
token = ""
|
||||||
|
if not re.fullmatch(r"[a-f0-9]{64}", token):
|
||||||
|
if not create:
|
||||||
|
return ""
|
||||||
|
token = secrets.token_hex(32)
|
||||||
|
secure = "; Secure" if getattr(request, "isSecure", lambda: False)() else ""
|
||||||
|
request.setHeader(b"Set-Cookie", f"ose_csrf={token}; Path=/; HttpOnly; SameSite=Lax{secure}".encode())
|
||||||
|
request._ose_csrf = token
|
||||||
|
return token
|
||||||
|
|
||||||
|
|
||||||
|
def protect_forms(request, markup):
|
||||||
|
token = csrf_token(request)
|
||||||
|
field = f'<input type="hidden" name="csrf" value="{token}">'
|
||||||
|
return re.sub(r'(<form\b[^>]*\bmethod=["\']post["\'][^>]*>)', lambda match: match[1] + field, markup, flags=re.I)
|
||||||
|
|
||||||
|
|
||||||
|
def valid_csrf(request, fields):
|
||||||
|
expected = csrf_token(request, create=False)
|
||||||
|
supplied = str(fields.get("csrf") or "")
|
||||||
|
return bool(expected and re.fullmatch(r"[a-f0-9]{64}", supplied) and secrets.compare_digest(expected, supplied))
|
||||||
Reference in new issue
Block a user
