dev/glo
3
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Update routes.py

Browse Source
This commit is contained in:
dev 2026-10-31 08:40:26 +00:00
parent 502836153d
commit 1cbdae8717
1 changed files with 36 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
routes.py
View File

@ -1102,35 +1102,19 @@ def register_routes(app, ctx):
if not csrf_valid(): if not csrf_valid():
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
user, error = require_writable_user() moderator, error = require_moderator()
if error: if error:
return error return error
limited = rate_limited("send") limited = rate_limited("admin")
if limited: if limited:
return limited return limited
with get_db() as db: cursor = db_write("DELETE FROM messages WHERE id = ?", (message_id,))
message = db.execute( if cursor.rowcount == 0:
"""
SELECT id, user_id
FROM messages
WHERE id = ?
""",
(message_id,),
).fetchone()
if not message:
return jsonify({"error": "Message not found."}), 404 return jsonify({"error": "Message not found."}), 404
is_owner = int(message["user_id"]) == int(user["id"]) audit("message_deleted", moderator["id"], "message", str(message_id))
is_moderator = user["role"] in {"moderator", "admin"}
if not is_owner and not is_moderator:
return jsonify({"error": "You cannot delete this message.", "code": "forbidden"}), 403
db_write("DELETE FROM messages WHERE id = ?", (message_id,))
audit("message_deleted", user["id"], "message", str(message_id))
return jsonify({"status": "deleted"}) return jsonify({"status": "deleted"})
@app.route("/api/messages/history") @app.route("/api/messages/history")
@ -3843,7 +3827,6 @@ def register_routes(app, ctx):
INSERT INTO pages ( INSERT INTO pages (
team_id, team_id,
channel_id, channel_id,
parent_page_id,
slug, slug,
title, title,
body, body,
@ -3854,7 +3837,7 @@ def register_routes(app, ctx):
created_at, created_at,
updated_at, updated_at,
is_active is_active
) VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1)
""", """,
( (
team["id"], team["id"],
@ -3878,7 +3861,7 @@ def register_routes(app, ctx):
with get_db() as db: with get_db() as db:
page = db.execute(f"{team_page_select()} WHERE p.id = ?", (page_id,)).fetchone() page = db.execute(f"{team_page_select()} WHERE p.id = ?", (page_id,)).fetchone()
return jsonify({"page": page_payload(ctx, page, user)}), 201 return jsonify({"page": page_payload(page, user)}), 201
@app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>") @app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>")
def api_get_page(team_slug: str, page_slug: str): def api_get_page(team_slug: str, page_slug: str):
@ -3989,7 +3972,7 @@ def register_routes(app, ctx):
with get_db() as db: with get_db() as db:
updated = db.execute(f"{team_page_select()} WHERE p.id = ?", (page["id"],)).fetchone() updated = db.execute(f"{team_page_select()} WHERE p.id = ?", (page["id"],)).fetchone()
return jsonify({"page": page_payload(ctx, updated, user)}) return jsonify({"page": page_payload(updated, user)})
@app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>", methods=["DELETE"]) @app.route("/api/teams/<string:team_slug>/pages/<string:page_slug>", methods=["DELETE"])
def api_delete_page(team_slug: str, page_slug: str): def api_delete_page(team_slug: str, page_slug: str):
@ -4053,8 +4036,7 @@ def register_routes(app, ctx):
role, role,
status, status,
restricted_until, restricted_until,
created_at, created_at
last_login_at
FROM users FROM users
WHERE id = ? WHERE id = ?
""", """,
@ -4099,6 +4081,7 @@ def register_routes(app, ctx):
@app.route("/api/profile", methods=["PATCH"]) @app.route("/api/profile", methods=["PATCH"])
def api_update_my_profile(): def api_update_my_profile():
if not csrf_valid(): if not csrf_valid():
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
@ -4121,8 +4104,6 @@ def register_routes(app, ctx):
bio_mode = str(data.get("bio_mode", "txt")).strip().lower() bio_mode = str(data.get("bio_mode", "txt")).strip().lower()
website = str(data.get("website", "")).strip() website = str(data.get("website", "")).strip()
location = str(data.get("location", "")).strip() location = str(data.get("location", "")).strip()
status_message = str(data.get("status_message", "")).strip()
banner_attachment_id = data.get("banner_attachment_id")
if len(display_name) > MAX_DISPLAY_NAME_LENGTH: if len(display_name) > MAX_DISPLAY_NAME_LENGTH:
return jsonify({"error": "Display name is too long."}), 400 return jsonify({"error": "Display name is too long."}), 400
@ -4136,18 +4117,6 @@ def register_routes(app, ctx):
return jsonify({"error": "Website is too long."}), 400 return jsonify({"error": "Website is too long."}), 400
if len(location) > MAX_LOCATION_LENGTH: if len(location) > MAX_LOCATION_LENGTH:
return jsonify({"error": "Location is too long."}), 400 return jsonify({"error": "Location is too long."}), 400
if len(status_message) > 160:
return jsonify({"error": "Status message is too long."}), 400
parsed_banner_attachment_id = None
if banner_attachment_id not in (None, ""):
parsed_banner_attachment_id, banner_error = resolve_owned_attachment(
get_db,
banner_attachment_id,
user["id"],
)
if banner_error:
return banner_error
db_write( db_write(
""" """
@ -4159,8 +4128,6 @@ def register_routes(app, ctx):
bio_mode = ?, bio_mode = ?,
website = ?, website = ?,
location = ?, location = ?,
status_message = ?,
banner_attachment_id = ?,
updated_at = ? updated_at = ?
WHERE user_id = ? WHERE user_id = ?
""", """,
@ -4171,8 +4138,6 @@ def register_routes(app, ctx):
bio_mode, bio_mode,
website, website,
location, location,
status_message,
parsed_banner_attachment_id,
now_iso(), now_iso(),
user["id"], user["id"],
), ),
@ -4187,8 +4152,7 @@ def register_routes(app, ctx):
role, role,
status, status,
restricted_until, restricted_until,
created_at, created_at
last_login_at
FROM users FROM users
WHERE id = ? WHERE id = ?
""", """,
@ -4214,6 +4178,7 @@ def register_routes(app, ctx):
(user["id"],), (user["id"],),
).fetchone() ).fetchone()
settings_row = db.execute( settings_row = db.execute(
""" """
SELECT SELECT
@ -4232,7 +4197,6 @@ def register_routes(app, ctx):
audit("profile_updated", user["id"], "user", str(user["id"])) audit("profile_updated", user["id"], "user", str(user["id"]))
return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)})
@app.route("/api/profiles/<string:username>") @app.route("/api/profiles/<string:username>")
def api_get_public_profile(username: str): def api_get_public_profile(username: str):
normalized = normalize_username(username) normalized = normalize_username(username)
@ -4246,8 +4210,7 @@ def register_routes(app, ctx):
role, role,
status, status,
restricted_until, restricted_until,
created_at, created_at
last_login_at
FROM users FROM users
WHERE username = ? WHERE username = ?
""", """,
@ -4293,7 +4256,6 @@ def register_routes(app, ctx):
return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())}) return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())})
@app.route("/api/settings") @app.route("/api/settings")
def api_get_settings(): def api_get_settings():
user, error = require_user() user, error = require_user()
@ -4308,7 +4270,6 @@ def register_routes(app, ctx):
name_color, name_color,
custom_css, custom_css,
theme_mode, theme_mode,
presence_mode,
profile_background_color, profile_background_color,
profile_banner_color, profile_banner_color,
profile_visibility, profile_visibility,
@ -4320,8 +4281,22 @@ def register_routes(app, ctx):
(user["id"],), (user["id"],),
).fetchone() ).fetchone()
return jsonify({"settings": dict(row)})
return jsonify(
{
"settings": {
"accent_color": row["accent_color"],
"name_color": row["name_color"],
"custom_css": row["custom_css"],
"theme_mode": row["theme_mode"],
"profile_background_color": row["profile_background_color"],
"profile_banner_color": row["profile_banner_color"],
"profile_visibility": row["profile_visibility"],
"last_active_visibility": row["last_active_visibility"],
"updated_at": row["updated_at"],
}
}
)
@app.route("/api/settings", methods=["PATCH"]) @app.route("/api/settings", methods=["PATCH"])
def api_update_settings(): def api_update_settings():
@ -4345,30 +4320,15 @@ def register_routes(app, ctx):
name_color = data.get("name_color") name_color = data.get("name_color")
custom_css = str(data.get("custom_css", "")) custom_css = str(data.get("custom_css", ""))
theme_mode = str(data.get("theme_mode", "dark")).strip().lower() theme_mode = str(data.get("theme_mode", "dark")).strip().lower()
presence_mode = str(data.get("presence_mode", "online")).strip().lower()
profile_background_color = data.get("profile_background_color")
profile_banner_color = data.get("profile_banner_color")
profile_visibility = str(data.get("profile_visibility", "public")).strip().lower()
last_active_visibility = str(data.get("last_active_visibility", "everyone")).strip().lower()
if not valid_hex_color(accent_color): if not valid_hex_color(accent_color):
return jsonify({"error": "Invalid accent color."}), 400 return jsonify({"error": "Invalid accent color."}), 400
if name_color not in (None, "") and not valid_hex_color(name_color): if name_color not in (None, "") and not valid_hex_color(name_color):
return jsonify({"error": "Invalid name color."}), 400 return jsonify({"error": "Invalid name color."}), 400
if profile_background_color not in (None, "") and not valid_hex_color(profile_background_color):
return jsonify({"error": "Invalid profile background color."}), 400
if profile_banner_color not in (None, "") and not valid_hex_color(profile_banner_color):
return jsonify({"error": "Invalid profile banner color."}), 400
if len(custom_css) > MAX_CUSTOM_CSS_LENGTH: if len(custom_css) > MAX_CUSTOM_CSS_LENGTH:
return jsonify({"error": "Custom CSS is too long."}), 400 return jsonify({"error": "Custom CSS is too long."}), 400
if theme_mode not in THEME_MODES: if theme_mode not in THEME_MODES:
return jsonify({"error": "Invalid theme mode."}), 400 return jsonify({"error": "Invalid theme mode."}), 400
if presence_mode not in {"online", "silent", "focus", "dnd"}:
return jsonify({"error": "Invalid presence mode."}), 400
if profile_visibility not in {"public", "members", "private"}:
return jsonify({"error": "Invalid profile visibility."}), 400
if last_active_visibility not in {"everyone", "members", "nobody"}:
return jsonify({"error": "Invalid last active visibility."}), 400
db_write( db_write(
""" """
@ -4378,7 +4338,6 @@ def register_routes(app, ctx):
name_color = ?, name_color = ?,
custom_css = ?, custom_css = ?,
theme_mode = ?, theme_mode = ?,
presence_mode = ?,
profile_background_color = ?, profile_background_color = ?,
profile_banner_color = ?, profile_banner_color = ?,
profile_visibility = ?, profile_visibility = ?,
@ -4391,11 +4350,10 @@ def register_routes(app, ctx):
normalized_hex_color(name_color) if name_color else None, normalized_hex_color(name_color) if name_color else None,
custom_css, custom_css,
theme_mode, theme_mode,
presence_mode, data.get("profile_background_color"),
normalized_hex_color(profile_background_color) if profile_background_color else None, data.get("profile_banner_color"),
normalized_hex_color(profile_banner_color) if profile_banner_color else None, str(data.get("profile_visibility", "public")).strip().lower(),
profile_visibility, str(data.get("last_active_visibility", "mutuals")).strip().lower(),
last_active_visibility,
now_iso(), now_iso(),
user["id"], user["id"],
), ),
@ -4409,7 +4367,6 @@ def register_routes(app, ctx):
name_color, name_color,
custom_css, custom_css,
theme_mode, theme_mode,
presence_mode,
profile_background_color, profile_background_color,
profile_banner_color, profile_banner_color,
profile_visibility, profile_visibility,
@ -5158,19 +5115,13 @@ def register_routes(app, ctx):
@app.route("/api/group-calls/<int:session_id>/leave", methods=["POST"]) @app.route("/api/group-calls/<int:session_id>/leave", methods=["POST"])
def api_leave_group_call_session(session_id: int): def api_leave_group_call_session(session_id: int):
if not csrf_valid():
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
user, error = require_user() user, error = require_user()
if error: if error:
return error return error
allow_without_csrf = (
request.headers.get("Sec-Fetch-Mode", "") == "no-cors"
or request.content_type == "text/plain;charset=UTF-8"
or request.content_type == "application/json"
)
if not csrf_valid() and not allow_without_csrf:
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
session_row = get_group_call_session_by_id(session_id) session_row = get_group_call_session_by_id(session_id)
if not session_row: if not session_row:
return jsonify({"error": "Group call session not found."}), 404 return jsonify({"error": "Group call session not found."}), 404
@ -5198,7 +5149,7 @@ def register_routes(app, ctx):
"participants": list_group_call_participants(session_id), "participants": list_group_call_participants(session_id),
} }
) )
@app.route("/api/group-calls/<int:session_id>/end", methods=["POST"]) @app.route("/api/group-calls/<int:session_id>/end", methods=["POST"])
def api_end_group_call_session(session_id: int): def api_end_group_call_session(session_id: int):
if not csrf_valid(): if not csrf_valid():