From 0c38c3bada91ad49f3a2af92e32c82b2b6d901be Mon Sep 17 00:00:00 2001 From: rekushi <1+rekushi@noreply.flipnote.world> Date: Thu, 29 Oct 2026 04:22:23 +0000 Subject: [PATCH] feat(web): integrate creator and administrator controls Connect the new management pages, protect POST forms, paginate comments and enforce hidden-content access rules. --- webapp.py | 158 ++++++++++++++++++++++++++++++------------------------ 1 file changed, 88 insertions(+), 70 deletions(-) diff --git a/webapp.py b/webapp.py index a9d2dc8..a1e5c7c 100644 --- a/webapp.py +++ b/webapp.py @@ -8,11 +8,18 @@ from twisted.web import resource from DB import Database from hatena import client_ip, request_args from reximemo import ppm_thumbnail_png +from server_config import load_config +from web_ui import page_number, pagination, protect_forms, valid_csrf +from web_features import WebFeatures, staff def _fields(request): + if request.args and any(key in request.args for key in (b"csrf", "csrf")): + return {(key.decode("utf-8", "replace") if isinstance(key, bytes) else key): + (values[-1].decode("utf-8", "replace") if isinstance(values[-1], bytes) else values[-1]) + for key, values in request.args.items() if values and key not in {b"logo", "logo"}} try: - raw = request.content.read().decode("utf-8", "replace") + raw = request.content.read(131073).decode("utf-8", "replace") except Exception: raw = "" return {k: v[-1] if v else "" for k, v in parse_qs(raw, keep_blank_values=True).items()} @@ -32,7 +39,7 @@ def _fmt_time(ts): return "" -class WebRoot(resource.Resource): +class WebRoot(WebFeatures, resource.Resource): isLeaf = True def __init__(self, docs_enabled=False): @@ -43,6 +50,7 @@ class WebRoot(resource.Resource): return Database.account_for_ip(client_ip(request)) def _page(self, request, title, body, *, account=None, active=""): + config = load_config() account = account if account is not None else self._account(request) browse_class = " active" if active == "browse" else "" home_class = " active" if active == "home" else "" @@ -53,6 +61,7 @@ class WebRoot(resource.Resource): nav_html = ( f'Home' f'Browse' + 'ChannelsSearch' f'About' ) if self.docs_enabled: @@ -62,24 +71,29 @@ class WebRoot(resource.Resource): if int(account.get("is_admin") or 0): actions += f'Admin' else: - actions = 'Sign inCreate account' + actions = 'Sign in' + if config["registration_enabled"]: + actions += 'Create account' - full_title = "RexiMemo OSE" if title == "Home" else f"{html.escape(title)} · RexiMemo OSE" + site_name = html.escape(config["site_name"]) + full_title = site_name if title == "Home" else f"{html.escape(title)} · {site_name}" docs_footer = 'Docs' if self.docs_enabled else "" page = f""" -{full_title} +{full_title}
-RexiMemo! Open Source Edition +{site_name}
{actions}
{body}
-""" +""" request.setHeader(b"content-type", b"text/html; charset=utf-8") - return page.encode("utf-8") + request.setHeader(b"Cache-Control", b"private, no-store") + request.setHeader(b"X-Content-Type-Options", b"nosniff") + return protect_forms(request, page).encode("utf-8") def _flipnote_card(self, fn): return ( - '
' + '
' '' '
%(title)s' '%(creator_name)s' @@ -92,17 +106,31 @@ class WebRoot(resource.Resource): "title": html.escape(fn.get("title") or "Untitled Flipnote"), "stars": int(fn.get("stars") or 0), "views": int(fn.get("views") or 0), + "hidden_class": " hidden-content" if fn.get("hidden") else "", } def render_GET(self, request): path = request.path.decode("utf-8", "replace") account = self._account(request) + include_hidden = staff(account) + extra = self._feature_get(request, path, account) + if extra is not None: + return extra + if path == "/branding/logo.png": + logo = Database.db_path.parent / "site-logo.png" + if not logo.is_file(): + request.setResponseCode(404) + return b"" + request.setHeader(b"Content-Type", b"image/png") + request.setHeader(b"Cache-Control", b"no-cache") + return logo.read_bytes() if path == "/": - rows = Database.list_flipnotes(limit=12) + rows = Database.list_flipnotes(limit=12, include_hidden=include_hidden) + config = load_config() body = ( '
' - 'Open Source Edition' - 'RexiMemo!

A small Flipnote Hatena-compatible server.

' + f'{html.escape(config["tagline"])}' + f'{html.escape(config["site_name"])}

{html.escape(config["welcome_message"])}

' '
Browse Flipnotes
' '

Recent Flipnotes

View all
' '
' + "".join(self._flipnote_card(r) for r in rows) + "
" @@ -137,31 +165,19 @@ class WebRoot(resource.Resource): from docs_page import render_docs return self._page(request, "Documentation", render_docs(), account=account, active="docs") - if path == "/browse": - sort = request_args(request).get("sort", ["new"])[0] - if sort not in {"new", "popular", "top"}: - sort = "new" - rows = Database.list_flipnotes(sort=sort, limit=100) - body = ( - '

Browse

Flipnotes posted to this server.

' - '
NewestPopularStars
' - '
' + "".join(self._flipnote_card(r) for r in rows) + "
" - ) - if not rows: - body += '

No Flipnotes found.

' - return self._page(request, "Browse", body, account=account, active="browse") if path.startswith("/thumb/") and path.endswith(".png"): public_id = path[len("/thumb/"):-4] - fn = Database.get_flipnote_by_public_id(public_id) + fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" try: - data = ppm_thumbnail_png(Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]), (320, 240)) + first = request_args(request).get("first", [""])[0] == "1" + data = ppm_thumbnail_png(Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden), (320, 240), first_frame=first) except Exception: request.setResponseCode(500); return b"" request.setHeader(b"content-type", b"image/png") - request.setHeader(b"Cache-Control", b"public, max-age=3600") + request.setHeader(b"Cache-Control", b"private, no-store") return data if path.startswith("/comment-thumb/") and path.endswith(".png"): @@ -170,7 +186,7 @@ class WebRoot(resource.Resource): comment_id = int(raw_id) except ValueError: request.setResponseCode(404); return b"" - raw = Database.comment_ppm(comment_id) + raw = Database.comment_ppm(comment_id, include_hidden=include_hidden) if raw is None: request.setResponseCode(404); return b"" try: @@ -178,41 +194,44 @@ class WebRoot(resource.Resource): except Exception: request.setResponseCode(500); return b"" request.setHeader(b"content-type", b"image/png") - request.setHeader(b"Cache-Control", b"public, max-age=3600") + request.setHeader(b"Cache-Control", b"private, no-store") return data if path.startswith("/media/") and path.endswith(".ppm"): public_id = path[len("/media/"):-4] - fn = Database.get_flipnote_by_public_id(public_id) + fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" request.setHeader(b"content-type", b"application/octet-stream") request.setHeader(b"content-disposition", b"inline") - request.setHeader(b"Cache-Control", b"public, max-age=3600") - return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]) + request.setHeader(b"Cache-Control", b"private, no-store") + return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden) if path.startswith("/flipnote/") and path.endswith(".ppm"): public_id = path[len("/flipnote/"):-4] - fn = Database.get_flipnote_by_public_id(public_id) + fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404); return b"" Database.AddDownload(fn["creator_fsid"], fn["filename"]) request.setHeader(b"content-type", b"application/octet-stream") request.setHeader(b"content-disposition", (f'attachment; filename="{fn["filename"]}.ppm"').encode("ascii", "ignore")) - return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"]) + request.setHeader(b"Cache-Control", b"private, no-store") + return Database.GetFlipnotePPM(fn["creator_fsid"], fn["filename"], include_hidden=include_hidden) if path.startswith("/watch/"): public_id = path[len("/watch/"):].strip("/") - fn = Database.get_flipnote_by_public_id(public_id) + fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) if not fn: request.setResponseCode(404) return self._page(request, "Not found", '

Flipnote not found

', account=account) Database.AddView(fn["creator_fsid"], fn["filename"]) - fn = Database.get_flipnote_by_public_id(public_id) - comments = Database.list_comments(fn["id"], limit=100) + fn = Database.get_flipnote_by_public_id(public_id, include_hidden=include_hidden) + comment_total = Database.count_comments(fn["id"], include_hidden=include_hidden) + comment_page = page_number(request_args(request).get("page", ["1"])[0], comment_total, 25) + comments = Database.list_comments(fn["id"], limit=25, offset=(comment_page-1)*25, include_hidden=include_hidden) channel = Database.get_channel(fn["channel_id"]) if fn.get("channel_id") else None room = Database.get_creator_room(fn["creator_room_id"]) - related = [r for r in Database.list_creator_flipnotes(fn["creator_room_id"], limit=8) if int(r["id"]) != int(fn["id"])][:4] + related = [r for r in Database.list_creator_flipnotes(fn["creator_room_id"], limit=8, include_hidden=include_hidden) if int(r["id"]) != int(fn["id"])][:4] try: ppm_size = Database.FlipnotePath(fn["creator_fsid"], fn["filename"]).stat().st_size size_text = "%.1f KiB" % (ppm_size / 1024.0) @@ -228,6 +247,10 @@ class WebRoot(resource.Resource): ) else: content = '
%s
' % html.escape(c.get("text_content") or "") + if c.get("hidden"): + content = '

Hidden by moderation. Only administrators can see this comment.

' + content + if staff(account) or Database.owns_comment(account["id"] if account else None, client_ip(request), c): + content += f'
' rendered_comments.append( '
%s%s
%s
' % ( html.escape(c["creator_public_id"], quote=True), html.escape(c["creator_name"]), _fmt_time(c["posted_at"]), content @@ -237,6 +260,9 @@ class WebRoot(resource.Resource): '
' '
' ) % html.escape(public_id, quote=True) + if not account and not load_config()["guest_posting"]: + comment_form = '

Sign in to leave a comment.

' + comment_form += pagination(comment_page, comment_total, 25, "/watch/" + public_id) related_html = "".join(self._flipnote_card(r) for r in related) body = ( '
Flipnote

%(title)s

' @@ -246,7 +272,7 @@ class WebRoot(resource.Resource): '
Flipnote playerPPM · flipnote.js
' '
' '
' - 'First frame of %(title_attr)s' + 'First frame of %(title_attr)s' '' '
Loading Flipnote…
' '
' @@ -279,7 +305,7 @@ class WebRoot(resource.Resource): "stars": int(fn.get("stars") or 0), "views": int(fn.get("views") or 0), "downloads": int(fn.get("downloads") or 0), - "comment_count": len(comments), + "comment_count": comment_total, "posted": _fmt_time(fn.get("posted_at")), "channel": html.escape((channel or {}).get("title") or "General"), "fsid": html.escape(fn.get("creator_fsid") or ""), @@ -292,22 +318,18 @@ class WebRoot(resource.Resource): % (html.escape(fn["creator_name"]), html.escape(fn["creator_public_id"], quote=True), related_html) ) if related_html else "", } + if fn.get("description"): + description = '
Description

' + html.escape(fn["description"]) + '

' + body = body.replace('
Manage Flipnote' + if fn.get("hidden"): + controls = '

This Flipnote is hidden. Only administrators can see it.

' + controls + if controls: + body = '
' + controls + '
' + body return self._page(request, fn.get("title") or "Flipnote", body, account=account) - if path.startswith("/creator/"): - public_id = path[len("/creator/"):].strip("/") - room = Database.get_creator_room_by_public_id(public_id) - if not room: - request.setResponseCode(404) - return self._page(request, "Not found", '

Creator not found

', account=account) - notes = Database.list_creator_flipnotes(room["id"], limit=100) - owner = "Account" if room.get("account_id") else "Guest/IP creator" - body = ( - '

%s

%s · Room created when this creator first posted a Flipnote.

' - '

Posted Flipnotes

%d
' - '
%s
' - ) % (html.escape(room["display_name"]), owner, len(notes), "".join(self._flipnote_card(r) for r in notes)) - return self._page(request, room["display_name"], body, account=account) if path == "/login": body = ( @@ -320,6 +342,8 @@ class WebRoot(resource.Resource): return self._page(request, "Sign in", body, account=account) if path == "/register": + if not load_config()["registration_enabled"]: + return self._error(request, "New account registration is currently closed.", 403) body = ( '
Create account
' '
' @@ -344,21 +368,6 @@ class WebRoot(resource.Resource): ) % (html.escape(account["username"]), room_link, html.escape(client_ip(request))) return self._page(request, "Account", body, account=account, active="account") - if path == "/admin": - if not account or not int(account.get("is_admin") or 0): - request.setResponseCode(403) - return self._page(request, "Admin", '

Admin access required

', account=account) - bans = Database.list_bans() - notes = Database.list_flipnotes(limit=100) - ban_rows = "".join('%s%s
' % (html.escape(b["ip"]), html.escape(b.get("reason") or ""), html.escape(b["ip"], quote=True)) for b in bans) - note_rows = "".join('%s%s
' % (html.escape(n["title"]), html.escape(n["creator_name"]), int(n["id"])) for n in notes) - body = ( - '

Admin

IP bans and Flipnote deletion.

' - '
Ban IP
' - '

Blocked IPs

%s
IPReason
' - '

Flipnotes

%s
TitleCreator
' - ) % (ban_rows, note_rows) - return self._page(request, "Admin", body, account=account, active="admin") request.setResponseCode(404) return self._page(request, "Not found", '

Not found

', account=account) @@ -368,6 +377,11 @@ class WebRoot(resource.Resource): fields = _fields(request) ip = client_ip(request) account = self._account(request) + if not valid_csrf(request, fields): + return self._error(request, "That form expired. Reload the page and try again.", 403) + extra = self._feature_post(request, path, fields, account) + if extra is not None: + return extra if path == "/login": found = Database.verify_account(fields.get("username", ""), fields.get("password", "")) @@ -378,6 +392,8 @@ class WebRoot(resource.Resource): return _redirect(request, "/account") if path == "/register": + if not load_config()["registration_enabled"]: + return self._error(request, "New account registration is currently closed.", 403) try: aid = Database.create_account(fields.get("username", ""), fields.get("password", "")) Database.login_ip(ip, aid) @@ -387,6 +403,8 @@ class WebRoot(resource.Resource): return _redirect(request, "/account") if path.startswith("/watch/") and path.endswith("/comment"): + if not account and not load_config()["guest_posting"]: + return self._error(request, "Sign in to leave a comment.", 403) public_id = path[len("/watch/"):-len("/comment")].strip("/") fn = Database.get_flipnote_by_public_id(public_id) if not fn: