RexiMemo/RexiMemo-OSE
1
1 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web): integrate creator and administrator controls

Browse Source 
Connect the new management pages, protect POST forms, paginate comments and enforce hidden-content access rules.
This commit is contained in:
rekushi committed 2026-10-29 04:22:23 +00:00
1 parent b776549445
commit 0c38c3bada
1 file changed
+88 -70
Download Patch File Download Diff File Expand all files Collapse all files
webapp.py
+88 -70
View File
@@ -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'<a class="nav-link{home_class}" href="/">Home</a>'
f'<a class="nav-link{browse_class}" href="/browse">Browse</a>'
'<a class="nav-link" href="/channels">Channels</a><a class="nav-link" href="/search">Search</a>'
f'<a class="nav-link{about_class}" href="/about">About</a>'
)
if self.docs_enabled:
@@ -62,24 +71,29 @@ class WebRoot(resource.Resource):
if int(account.get("is_admin") or 0):
actions += f'<a class="nav-action{admin_class}" href="/admin">Admin</a>'
else:
actions = '<a class="nav-action ghost" href="/login">Sign in</a><a class="nav-action" href="/register">Create account</a>'
actions = '<a class="nav-action ghost" href="/login">Sign in</a>'
if config["registration_enabled"]:
actions += '<a class="nav-action" href="/register">Create account</a>'
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 = '<a href="/docs">Docs</a>' if self.docs_enabled else ""
page = f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>{full_title}</title><meta name="theme-color" content="#b13b29"><link rel="stylesheet" href="/static/site.css"></head>
<title>{full_title}</title><meta name="theme-color" content="{config['accent_colour']}"><link rel="stylesheet" href="/static/site.css?v=2"><style>:root{{--accent:{config['accent_colour']};--accent-hover:{config['accent_colour']}}}</style></head>
<body><header class="site-header"><div class="header-inner">
<a class="brand" href="/" aria-label="RexiMemo Open Source Edition"><img src="/static/RexiMemo_OSE_Logo.png" alt="RexiMemo! Open Source Edition"></a>
<a class="brand" href="/" aria-label="{site_name}"><img src="{html.escape(config['logo_url'], quote=True)}" alt="{site_name}"></a>
<nav class="main-nav">{nav_html}</nav><div class="header-actions">{actions}</div>
</div></header><main class="page-shell">{body}</main>
<footer class="site-footer"><div><strong>RexiMemo: Open Source Edition</strong><span>AGPL-3.0</span></div><div><a href="/browse">Browse Flipnotes</a><a href="/about">About</a>{docs_footer}</div></footer></body></html>"""
<footer class="site-footer"><div><strong>{site_name}</strong><span>Powered by RexiMemo OSE · AGPL-3.0</span></div><div><a href="/browse">Browse Flipnotes</a><a href="/about">About</a>{docs_footer}</div></footer></body></html>"""
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 (
'<article class="flipnote-card">'
'<article class="flipnote-card%(hidden_class)s">'
'<a class="flipnote-thumb" href="/watch/%(public)s"><img loading="lazy" src="/thumb/%(public)s.png" alt=""></a>'
'<div class="flipnote-card-body"><a class="flipnote-title" href="/watch/%(public)s">%(title)s</a>'
'<a class="flipnote-creator" href="/creator/%(creator)s">%(creator_name)s</a>'
@@ -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 = (
'<section class="home-intro-card"><div class="home-intro-main">'
'<span class="home-intro-eyebrow">Open Source Edition</span>'
'<strong>RexiMemo!</strong><p>A small Flipnote Hatena-compatible server.</p></div>'
f'<span class="home-intro-eyebrow">{html.escape(config["tagline"])}</span>'
f'<strong>{html.escape(config["site_name"])}</strong><p>{html.escape(config["welcome_message"])}</p></div>'
'<div class="home-intro-actions"><a class="button primary" href="/browse">Browse Flipnotes</a></div></section>'
'<section class="section first-section"><div class="section-head"><h1>Recent Flipnotes</h1><a href="/browse">View all</a></div>'
'<div class="flipnote-grid">' + "".join(self._flipnote_card(r) for r in rows) + "</div></section>"
@@ -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 = (
'<header class="page-head split"><div><h1>Browse</h1><p>Flipnotes posted to this server.</p></div>'
'<div class="tabs"><a href="/browse?sort=new">Newest</a><a href="/browse?sort=popular">Popular</a><a href="/browse?sort=top">Stars</a></div></header>'
'<div class="flipnote-grid">' + "".join(self._flipnote_card(r) for r in rows) + "</div>"
)
if not rows:
body += '<p class="empty-state">No Flipnotes found.</p>'
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", '<header class="page-head"><h1>Flipnote not found</h1></header>', 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 = '<div class="comment-text">%s</div>' % html.escape(c.get("text_content") or "")
if c.get("hidden"):
content = '<p class="moderation-note">Hidden by moderation. Only administrators can see this comment.</p>' + content
if staff(account) or Database.owns_comment(account["id"] if account else None, client_ip(request), c):
content += f'<form class="comment-tools" method="post" action="/comments/{c["id"]}/delete"><input type="hidden" name="flipnote" value="{html.escape(public_id, quote=True)}"><button class="button small subtle">Remove comment</button></form>'
rendered_comments.append(
'<div class="comment"><div class="comment-author"><a href="/creator/%s">%s</a><small>%s</small></div><div class="comment-content">%s</div></div>' % (
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):
'<form method="post" action="/watch/%s/comment" class="comment-form">'
'<input name="text" maxlength="500" placeholder="Write a comment"><button class="button primary">Post comment</button></form>'
) % html.escape(public_id, quote=True)
if not account and not load_config()["guest_posting"]:
comment_form = '<p class="notice"><a href="/login">Sign in</a> to leave a comment.</p>'
comment_form += pagination(comment_page, comment_total, 25, "/watch/" + public_id)
related_html = "".join(self._flipnote_card(r) for r in related)
body = (
'<header class="watch-head"><div><div class="watch-kicker">Flipnote</div><h1>%(title)s</h1>'
@@ -246,7 +272,7 @@ class WebRoot(resource.Resource):
'<div class="preview-bar"><span>Flipnote player</span><span>PPM · flipnote.js</span></div>'
'<div class="preview-stage flipnote-stage" data-flipnote-player data-source="/media/%(public)s.ppm">'
'<div class="flipnote-player-canvas" aria-label="Flipnote playback canvas"></div>'
'<img class="flipnote-player-fallback" data-fallback-label="First-frame preview" src="/thumb/%(public)s.png" alt="First frame of %(title_attr)s">'
'<img class="flipnote-player-fallback" data-fallback-label="First-frame preview" src="/thumb/%(public)s.png?first=1" alt="First frame of %(title_attr)s">'
'<button class="flipnote-big-play" type="button" data-action="big-play" aria-label="Play Flipnote" hidden>▶</button>'
'<div class="flipnote-player-status" data-display="status" role="status">Loading Flipnote…</div></div>'
'<div class="flipnote-controls" aria-label="Flipnote playback controls">'
@@ -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 = '<section class="panel description-panel"><div class="panel-title">Description</div><p>' + html.escape(fn["description"]) + '</p></section>'
body = body.replace('<section class="comments-panel', description + '<section class="comments-panel', 1)
controls = ''
if staff(account) or Database.owns_flipnote(account["id"] if account else None, client_ip(request), fn):
controls = f'<a class="button" href="/watch/{html.escape(public_id, quote=True)}/manage">Manage Flipnote</a>'
if fn.get("hidden"):
controls = '<p class="moderation-note">This Flipnote is hidden. Only administrators can see it.</p>' + controls
if controls:
body = '<div class="watch-management">' + controls + '</div>' + 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", '<header class="page-head"><h1>Creator not found</h1></header>', account=account)
notes = Database.list_creator_flipnotes(room["id"], limit=100)
owner = "Account" if room.get("account_id") else "Guest/IP creator"
body = (
'<header class="page-head creator-head"><div><h1>%s</h1><p>%s · Room created when this creator first posted a Flipnote.</p></div></header>'
'<section class="section first-section"><div class="section-head"><h2>Posted Flipnotes</h2><span>%d</span></div>'
'<div class="flipnote-grid">%s</div></section>'
) % (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 = (
'<div class="auth-wrap"><section class="auth-card"><div class="panel-title">Create account</div><div class="auth-body">'
'<form method="post" class="stacked-form"><label>Username<input name="username" maxlength="24"></label>'
@@ -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", '<header class="page-head"><h1>Admin access required</h1></header>', account=account)
bans = Database.list_bans()
notes = Database.list_flipnotes(limit=100)
ban_rows = "".join('<tr><td><code>%s</code></td><td>%s</td><td><form method="post" action="/admin/unban"><input type="hidden" name="ip" value="%s"><button class="button small">Unban</button></form></td></tr>' % (html.escape(b["ip"]), html.escape(b.get("reason") or ""), html.escape(b["ip"], quote=True)) for b in bans)
note_rows = "".join('<tr><td>%s</td><td>%s</td><td><form method="post" action="/admin/delete"><input type="hidden" name="id" value="%d"><button class="button small danger">Delete</button></form></td></tr>' % (html.escape(n["title"]), html.escape(n["creator_name"]), int(n["id"])) for n in notes)
body = (
'<header class="page-head"><div><h1>Admin</h1><p>IP bans and Flipnote deletion.</p></div></header>'
'<section class="panel"><div class="panel-title">Ban IP</div><form method="post" action="/admin/ban" class="admin-ban-form"><input name="ip" placeholder="IP address"><input name="reason" placeholder="Reason"><button class="button primary">Ban</button></form></section>'
'<section class="section"><div class="section-head"><h2>Blocked IPs</h2></div><div class="table-wrap"><table><tr><th>IP</th><th>Reason</th><th></th></tr>%s</table></div></section>'
'<section class="section"><div class="section-head"><h2>Flipnotes</h2></div><div class="table-wrap"><table><tr><th>Title</th><th>Creator</th><th></th></tr>%s</table></div></section>'
) % (ban_rows, note_rows)
return self._page(request, "Admin", body, account=account, active="admin")
request.setResponseCode(404)
return self._page(request, "Not found", '<header class="page-head"><h1>Not found</h1></header>', 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: