519 lines
18 KiB
Python
519 lines
18 KiB
Python
from .common import (
|
|
MAX_CUSTOM_CSS_LENGTH,
|
|
MAX_DISPLAY_NAME_LENGTH,
|
|
MAX_LOCATION_LENGTH,
|
|
MAX_MESSAGE_LENGTH,
|
|
MAX_PROFILE_BADGE_LENGTH,
|
|
MAX_PROFILE_BIO_LENGTH,
|
|
MAX_WEBSITE_LENGTH,
|
|
DM_LAYOUT_MODES,
|
|
LAST_ACTIVE_VISIBILITIES,
|
|
PRESENCE_MODES,
|
|
PROFILE_BIO_MODES,
|
|
PROFILE_VISIBILITIES,
|
|
THEME_MODES,
|
|
now_iso,
|
|
csrf_valid,
|
|
rate_limited,
|
|
require_user,
|
|
require_writable_user,
|
|
normalize_username,
|
|
audit,
|
|
user_payload,
|
|
safe_json,
|
|
jsonify,
|
|
get_db,
|
|
db_write,
|
|
profile_payload,
|
|
can_view_profile_record,
|
|
valid_hex_color,
|
|
normalized_hex_color,
|
|
)
|
|
|
|
|
|
def register_profile_routes(app):
|
|
@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,
|
|
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,
|
|
presence_mode,
|
|
profile_background_color,
|
|
profile_banner_color,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
dm_layout_mode,
|
|
show_presence_badge
|
|
FROM user_settings
|
|
WHERE user_id = ?
|
|
""",
|
|
(user["id"],),
|
|
).fetchone()
|
|
|
|
return jsonify({"profile": profile_payload(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()
|
|
status_message = str(data.get("status_message", "")).strip()
|
|
|
|
banner_attachment_id = data.get("banner_attachment_id")
|
|
if banner_attachment_id not in (None, ""):
|
|
try:
|
|
banner_attachment_id = int(banner_attachment_id)
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "Invalid banner attachment."}), 400
|
|
else:
|
|
banner_attachment_id = None
|
|
|
|
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) > 160:
|
|
return jsonify({"error": "Status message is too long."}), 400
|
|
|
|
if banner_attachment_id is not None:
|
|
with get_db() as db:
|
|
attachment = db.execute(
|
|
"""
|
|
SELECT id, uploader_user_id
|
|
FROM attachments
|
|
WHERE id = ?
|
|
""",
|
|
(banner_attachment_id,),
|
|
).fetchone()
|
|
if not attachment:
|
|
return jsonify({"error": "Banner attachment not found."}), 404
|
|
if attachment["uploader_user_id"] != user["id"]:
|
|
return jsonify({"error": "You can only use your own uploaded image."}), 403
|
|
|
|
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,
|
|
presence_mode,
|
|
profile_background_color,
|
|
profile_banner_color,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
dm_layout_mode,
|
|
show_presence_badge
|
|
FROM user_settings
|
|
WHERE user_id = ?
|
|
""",
|
|
(user["id"],),
|
|
).fetchone()
|
|
|
|
audit("profile_updated", user["id"], "user", str(user["id"]))
|
|
return jsonify({"profile": profile_payload(user_payload, user_row, profile_row, settings_row, user)})
|
|
|
|
@app.route("/api/profiles/<string:username>")
|
|
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,
|
|
presence_mode,
|
|
profile_background_color,
|
|
profile_banner_color,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
dm_layout_mode,
|
|
show_presence_badge
|
|
FROM user_settings
|
|
WHERE user_id = ?
|
|
""",
|
|
(user_row["id"],),
|
|
).fetchone()
|
|
|
|
if not can_view_profile_record(settings_row, viewer, user_row["id"]):
|
|
if not viewer:
|
|
return jsonify({"error": "Login required.", "code": "login_required"}), 401
|
|
return jsonify({"error": "You do not have access to this profile.", "code": "forbidden"}), 403
|
|
|
|
return jsonify({"profile": profile_payload(user_payload, user_row, profile_row, settings_row, viewer)})
|
|
|
|
@app.route("/api/settings")
|
|
def api_get_settings():
|
|
user, error = require_user()
|
|
if error:
|
|
return error
|
|
|
|
with get_db() as db:
|
|
row = db.execute(
|
|
"""
|
|
SELECT
|
|
accent_color,
|
|
name_color,
|
|
custom_css,
|
|
theme_mode,
|
|
profile_background_color,
|
|
profile_banner_color,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
presence_mode,
|
|
dm_layout_mode,
|
|
show_presence_badge,
|
|
updated_at
|
|
FROM user_settings
|
|
WHERE user_id = ?
|
|
""",
|
|
(user["id"],),
|
|
).fetchone()
|
|
|
|
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"],
|
|
"presence_mode": row["presence_mode"],
|
|
"dm_layout_mode": row["dm_layout_mode"],
|
|
"show_presence_badge": bool(row["show_presence_badge"]),
|
|
"updated_at": row["updated_at"],
|
|
}
|
|
}
|
|
)
|
|
|
|
@app.route("/api/settings", methods=["PATCH"])
|
|
def api_update_settings():
|
|
if not csrf_valid():
|
|
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
|
|
|
|
limited = rate_limited("settings")
|
|
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
|
|
|
|
accent_color = data.get("accent_color", "#f7dce8")
|
|
name_color = data.get("name_color")
|
|
custom_css = str(data.get("custom_css", ""))
|
|
theme_mode = str(data.get("theme_mode", "dark")).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()
|
|
presence_mode = str(data.get("presence_mode", "online")).strip().lower()
|
|
dm_layout_mode = str(data.get("dm_layout_mode", "cards")).strip().lower()
|
|
show_presence_badge_raw = data.get("show_presence_badge", True)
|
|
|
|
if not valid_hex_color(accent_color):
|
|
return jsonify({"error": "Invalid accent color."}), 400
|
|
if name_color not in (None, "") and not valid_hex_color(name_color):
|
|
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:
|
|
return jsonify({"error": "Custom CSS is too long."}), 400
|
|
if theme_mode not in THEME_MODES:
|
|
return jsonify({"error": "Invalid theme mode."}), 400
|
|
if profile_visibility not in PROFILE_VISIBILITIES:
|
|
return jsonify({"error": "Invalid profile visibility."}), 400
|
|
if last_active_visibility not in LAST_ACTIVE_VISIBILITIES:
|
|
return jsonify({"error": "Invalid last active visibility."}), 400
|
|
if presence_mode not in PRESENCE_MODES:
|
|
return jsonify({"error": "Invalid presence mode."}), 400
|
|
if dm_layout_mode not in DM_LAYOUT_MODES:
|
|
return jsonify({"error": "Invalid dm layout mode."}), 400
|
|
|
|
show_presence_badge = 1 if bool(show_presence_badge_raw) else 0
|
|
|
|
db_write(
|
|
"""
|
|
UPDATE user_settings
|
|
SET
|
|
accent_color = ?,
|
|
name_color = ?,
|
|
custom_css = ?,
|
|
theme_mode = ?,
|
|
profile_background_color = ?,
|
|
profile_banner_color = ?,
|
|
profile_visibility = ?,
|
|
last_active_visibility = ?,
|
|
presence_mode = ?,
|
|
dm_layout_mode = ?,
|
|
show_presence_badge = ?,
|
|
updated_at = ?
|
|
WHERE user_id = ?
|
|
""",
|
|
(
|
|
str(accent_color).lower(),
|
|
normalized_hex_color(name_color) if name_color else None,
|
|
custom_css,
|
|
theme_mode,
|
|
normalized_hex_color(profile_background_color) if profile_background_color else None,
|
|
normalized_hex_color(profile_banner_color) if profile_banner_color else None,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
presence_mode,
|
|
dm_layout_mode,
|
|
show_presence_badge,
|
|
now_iso(),
|
|
user["id"],
|
|
),
|
|
)
|
|
|
|
with get_db() as db:
|
|
row = db.execute(
|
|
"""
|
|
SELECT
|
|
accent_color,
|
|
name_color,
|
|
custom_css,
|
|
theme_mode,
|
|
profile_background_color,
|
|
profile_banner_color,
|
|
profile_visibility,
|
|
last_active_visibility,
|
|
presence_mode,
|
|
dm_layout_mode,
|
|
show_presence_badge,
|
|
updated_at
|
|
FROM user_settings
|
|
WHERE user_id = ?
|
|
""",
|
|
(user["id"],),
|
|
).fetchone()
|
|
|
|
audit("settings_updated", user["id"], "user", str(user["id"]))
|
|
return jsonify(
|
|
{
|
|
"settings": {
|
|
**dict(row),
|
|
"show_presence_badge": bool(row["show_presence_badge"]),
|
|
}
|
|
}
|
|
)
|
|
|
|
@app.route("/api/scheduled-messages", methods=["POST"])
|
|
def api_create_scheduled_message():
|
|
if not csrf_valid():
|
|
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
|
|
|
|
user, error = require_writable_user()
|
|
if error:
|
|
return error
|
|
|
|
try:
|
|
data = safe_json()
|
|
except ValueError as error:
|
|
return jsonify({"error": str(error)}), 400
|
|
|
|
text = str(data.get("text", "")).strip()
|
|
deliver_at = str(data.get("deliver_at", "")).strip()
|
|
scope_kind = str(data.get("scope_kind", "")).strip()
|
|
scope_id = data.get("scope_id")
|
|
|
|
if not text:
|
|
return jsonify({"error": "Scheduled message text is required."}), 400
|
|
if len(text) > MAX_MESSAGE_LENGTH:
|
|
return jsonify({"error": "Scheduled message is too long."}), 400
|
|
if not deliver_at:
|
|
return jsonify({"error": "deliver_at is required."}), 400
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "stubbed",
|
|
"scheduled_message": {
|
|
"id": 0,
|
|
"text": text,
|
|
"deliver_at": deliver_at,
|
|
"scope_kind": scope_kind,
|
|
"scope_id": scope_id,
|
|
}
|
|
}
|
|
), 201
|