diff --git a/routes.py b/routes.py index de37e6d..1c34708 100644 --- a/routes.py +++ b/routes.py @@ -4603,197 +4603,322 @@ def register_routes(app, ctx): payload["versions"] = [] return jsonify({"page": payload, "shared": True}) - @app.route("/api/profile") - def api_get_my_profile(): - user, error = require_user() - if error: - return error +@app.route("/api/profile") +def api_get_my_profile(): + user, error = require_user() + if error: + return error - with get_db() as db: - user_row = db.execute( - """ - SELECT - id, - username, - role, - status, - restricted_until, - created_at - FROM users - WHERE id = ? - """, - (user["id"],), - ).fetchone() - profile_row = db.execute( - """ - SELECT - user_id, - display_name, - badge_text, - bio, - bio_mode, - website, - location, - updated_at - FROM user_profiles - WHERE user_id = ? - """, - (user["id"],), - ).fetchone() - settings_row = db.execute( - """ - SELECT - name_color - FROM user_settings - WHERE user_id = ? - """, - (user["id"],), - ).fetchone() - - return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, user)}) - - @app.route("/api/profile", methods=["PATCH"]) - def api_update_my_profile(): - if not csrf_valid(): - return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 - limited = rate_limited("profile") - if limited: - return limited - user, error = require_writable_user() - if error: - return error - - try: - data = safe_json() - except ValueError as error: - return jsonify({"error": str(error)}), 400 - - display_name = str(data.get("display_name", "")).strip() - badge_text = str(data.get("badge_text", "")).strip() - bio = str(data.get("bio", "")).strip() - bio_mode = str(data.get("bio_mode", "txt")).strip().lower() - website = str(data.get("website", "")).strip() - location = str(data.get("location", "")).strip() - - if len(display_name) > MAX_DISPLAY_NAME_LENGTH: - return jsonify({"error": "Display name is too long."}), 400 - if len(badge_text) > MAX_PROFILE_BADGE_LENGTH: - return jsonify({"error": "Badge text is too long."}), 400 - if len(bio) > MAX_PROFILE_BIO_LENGTH: - return jsonify({"error": "Bio is too long."}), 400 - if bio_mode not in PROFILE_BIO_MODES: - return jsonify({"error": "Invalid bio mode."}), 400 - if len(website) > MAX_WEBSITE_LENGTH: - return jsonify({"error": "Website is too long."}), 400 - if len(location) > MAX_LOCATION_LENGTH: - return jsonify({"error": "Location is too long."}), 400 - - db_write( + with get_db() as db: + user_row = db.execute( """ - UPDATE user_profiles - SET - display_name = ?, - badge_text = ?, - bio = ?, - bio_mode = ?, - website = ?, - location = ?, - updated_at = ? - WHERE user_id = ? + SELECT + id, + username, + role, + status, + restricted_until, + created_at, + last_login_at + FROM users + WHERE id = ? """, - ( - display_name or user["username"], + (user["id"],), + ).fetchone() + + profile_row = db.execute( + """ + SELECT + user_id, + display_name, badge_text, bio, bio_mode, website, location, - now_iso(), - user["id"], - ), + status_message, + banner_attachment_id, + updated_at + FROM user_profiles + WHERE user_id = ? + """, + (user["id"],), + ).fetchone() + + settings_row = db.execute( + """ + SELECT + name_color, + profile_background_color, + profile_banner_color, + profile_visibility, + last_active_visibility, + presence_mode + FROM user_settings + WHERE user_id = ? + """, + (user["id"],), + ).fetchone() + + return jsonify({ + "profile": profile_payload( + html_module, + user_payload, + user_row, + profile_row, + settings_row, + user, ) + }) + + +@app.route("/api/profile", methods=["PATCH"]) +def api_update_my_profile(): + if not csrf_valid(): + return jsonify({"error": "Security token expired.", "code": "csrf"}), 403 + + limited = rate_limited("profile") + if limited: + return limited + + user, error = require_writable_user() + if error: + return error + + try: + data = safe_json() + except ValueError as exc: + return jsonify({"error": str(exc)}), 400 + + display_name = str(data.get("display_name", "")).strip() + badge_text = str(data.get("badge_text", "")).strip() + bio = str(data.get("bio", "")).strip() + bio_mode = str(data.get("bio_mode", "txt")).strip().lower() + website = str(data.get("website", "")).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: + return jsonify({"error": "Display name is too long."}), 400 + if len(badge_text) > MAX_PROFILE_BADGE_LENGTH: + return jsonify({"error": "Badge text is too long."}), 400 + if len(bio) > MAX_PROFILE_BIO_LENGTH: + return jsonify({"error": "Bio is too long."}), 400 + if bio_mode not in PROFILE_BIO_MODES: + return jsonify({"error": "Invalid bio mode."}), 400 + if len(website) > MAX_WEBSITE_LENGTH: + return jsonify({"error": "Website is too long."}), 400 + if len(location) > MAX_LOCATION_LENGTH: + return jsonify({"error": "Location is too long."}), 400 + if len(status_message) > constants["MAX_STATUS_MESSAGE_LENGTH"]: + return jsonify({"error": "Status message is too long."}), 400 + + if banner_attachment_id in ("", None): + banner_attachment_id = None + else: + try: + banner_attachment_id = int(banner_attachment_id) + except (TypeError, ValueError): + return jsonify({"error": "Invalid banner attachment."}), 400 with get_db() as db: - user_row = db.execute( + attachment = db.execute( """ SELECT id, - username, - role, - status, - restricted_until, - created_at - FROM users + uploader_user_id + FROM attachments WHERE id = ? """, - (user["id"],), - ).fetchone() - profile_row = db.execute( - """ - SELECT - user_id, - display_name, - badge_text, - bio, - bio_mode, - website, - location, - updated_at - FROM user_profiles - WHERE user_id = ? - """, - (user["id"],), - ).fetchone() - settings_row = db.execute( - "SELECT name_color FROM user_settings WHERE user_id = ?", - (user["id"],), + (banner_attachment_id,), ).fetchone() - 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)}) + if not attachment: + return jsonify({"error": "Banner attachment not found."}), 404 + if int(attachment["uploader_user_id"]) != int(user["id"]): + return jsonify({"error": "You can only use your own uploaded image as banner."}), 403 - @app.route("/api/profiles/") - def api_get_public_profile(username: str): - normalized = normalize_username(username) - with get_db() as db: - user_row = db.execute( - """ - SELECT - id, - username, - role, - status, - restricted_until, - created_at - FROM users - WHERE username = ? - """, - (normalized,), - ).fetchone() - if not user_row: - return jsonify({"error": "Profile not found."}), 404 - profile_row = db.execute( - """ - SELECT - user_id, - display_name, - badge_text, - bio, - bio_mode, - website, - location, - updated_at - FROM user_profiles - WHERE user_id = ? - """, - (user_row["id"],), - ).fetchone() - settings_row = db.execute( - "SELECT name_color FROM user_settings WHERE user_id = ?", - (user_row["id"],), - ).fetchone() - return jsonify({"profile": profile_payload(html_module, user_payload, user_row, profile_row, settings_row, current_user())}) + db_write( + """ + UPDATE user_profiles + SET + display_name = ?, + badge_text = ?, + bio = ?, + bio_mode = ?, + website = ?, + location = ?, + status_message = ?, + banner_attachment_id = ?, + updated_at = ? + WHERE user_id = ? + """, + ( + display_name or user["username"], + badge_text, + bio, + bio_mode, + website, + location, + status_message, + banner_attachment_id, + now_iso(), + user["id"], + ), + ) + with get_db() as db: + user_row = db.execute( + """ + SELECT + id, + username, + role, + status, + restricted_until, + created_at, + last_login_at + FROM users + WHERE id = ? + """, + (user["id"],), + ).fetchone() + + profile_row = db.execute( + """ + SELECT + user_id, + display_name, + badge_text, + bio, + bio_mode, + website, + location, + status_message, + banner_attachment_id, + updated_at + FROM user_profiles + WHERE user_id = ? + """, + (user["id"],), + ).fetchone() + + settings_row = db.execute( + """ + SELECT + name_color, + profile_background_color, + profile_banner_color, + profile_visibility, + last_active_visibility, + presence_mode + FROM user_settings + WHERE user_id = ? + """, + (user["id"],), + ).fetchone() + + 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, + ) + }) + + +@app.route("/api/profiles/") +def api_get_public_profile(username: str): + normalized = normalize_username(username) + viewer = current_user() + + with get_db() as db: + user_row = db.execute( + """ + SELECT + id, + username, + role, + status, + restricted_until, + created_at, + last_login_at + FROM users + WHERE username = ? + """, + (normalized,), + ).fetchone() + + if not user_row: + return jsonify({"error": "Profile not found."}), 404 + + profile_row = db.execute( + """ + SELECT + user_id, + display_name, + badge_text, + bio, + bio_mode, + website, + location, + status_message, + banner_attachment_id, + updated_at + FROM user_profiles + WHERE user_id = ? + """, + (user_row["id"],), + ).fetchone() + + settings_row = db.execute( + """ + SELECT + name_color, + profile_background_color, + profile_banner_color, + profile_visibility, + last_active_visibility, + presence_mode + FROM user_settings + WHERE user_id = ? + """, + (user_row["id"],), + ).fetchone() + + profile_visibility = ( + settings_row["profile_visibility"] + if settings_row and "profile_visibility" in settings_row.keys() + else "public" + ) + + is_self = bool(viewer and int(viewer["id"]) == int(user_row["id"])) + + if profile_visibility == "private": + if not is_self and not (viewer and viewer["role"] in {"moderator", "admin"}): + return jsonify({"error": "Profile not available."}), 403 + + if profile_visibility == "members": + if not is_self and not viewer: + return jsonify({"error": "Profile not available."}), 403 + + return jsonify({ + "profile": profile_payload( + html_module, + user_payload, + user_row, + profile_row, + settings_row, + viewer, + ) + }) + @app.route("/api/settings") def api_get_settings(): user, error = require_user()