414 lines
12 KiB
Python
414 lines
12 KiB
Python
import json
|
|
|
|
from flask import jsonify, request
|
|
|
|
from config import (
|
|
CALLS_BASE,
|
|
CALL_HELPER_SHARED_SECRET,
|
|
CALL_TOKEN_TTL_SECONDS,
|
|
CONSTANTS,
|
|
PRIVATE_ACCESS_KEY,
|
|
UPLOAD_DIR,
|
|
now_iso,
|
|
parse_iso,
|
|
utc_now,
|
|
)
|
|
from db import get_db, db_write
|
|
from auth import (
|
|
safe_json,
|
|
csrf_valid,
|
|
rate_limited,
|
|
current_user,
|
|
require_user,
|
|
require_writable_user,
|
|
require_moderator,
|
|
require_admin,
|
|
normalize_username,
|
|
audit,
|
|
hash_identifier,
|
|
sign_call_helper_token,
|
|
user_payload,
|
|
)
|
|
from models import (
|
|
ROOM_SLUG_RE,
|
|
MESSAGE_SELECT,
|
|
team_page_select,
|
|
team_slug_from_name,
|
|
page_slug_from_title,
|
|
room_by_id,
|
|
get_room_by_slug,
|
|
can_access_room,
|
|
can_write_room,
|
|
can_manage_room,
|
|
room_payload,
|
|
get_team_by_slug,
|
|
get_channel_by_id,
|
|
get_channel_by_slug,
|
|
get_channel_by_legacy_room_id,
|
|
team_member_role,
|
|
is_team_member,
|
|
is_team_admin,
|
|
can_edit_team,
|
|
can_access_team,
|
|
can_access_channel,
|
|
can_write_channel,
|
|
team_payload,
|
|
channel_payload,
|
|
get_dm_for_user,
|
|
get_dm_other_user,
|
|
validate_dm_access,
|
|
dm_payload,
|
|
find_or_create_dm,
|
|
update_presence,
|
|
room_presence_count,
|
|
channel_presence_count,
|
|
dm_presence_count,
|
|
)
|
|
from attachments import (
|
|
attachment_list_for_message,
|
|
attachment_payload_legacy_from_message_select_row,
|
|
resolve_owned_attachment,
|
|
resolve_owned_attachment_ids,
|
|
)
|
|
from profiles import (
|
|
DM_LAYOUT_MODES,
|
|
LAST_ACTIVE_VISIBILITIES,
|
|
PAGE_TYPES,
|
|
PAGE_VISIBILITIES,
|
|
PRESENCE_MODES,
|
|
PROFILE_BIO_MODES,
|
|
PROFILE_VISIBILITIES,
|
|
REACTION_EMOJI_RE,
|
|
TEAM_MEMBER_ROLES,
|
|
THEME_MODES,
|
|
normalized_hex_color,
|
|
valid_hex_color,
|
|
resolve_name_color,
|
|
page_payload,
|
|
profile_payload,
|
|
reaction_summary_for_message,
|
|
list_dms_for_user,
|
|
richer_message_payload,
|
|
feed_message_payload,
|
|
can_view_profile_record,
|
|
)
|
|
from calls import (
|
|
append_call_signal,
|
|
call_payload,
|
|
create_call_session,
|
|
end_call,
|
|
get_call_with_dm,
|
|
list_call_signals,
|
|
build_room_helper_call_id,
|
|
build_channel_helper_call_id,
|
|
get_or_create_group_call_session,
|
|
group_call_session_payload,
|
|
get_group_call_session_by_id,
|
|
upsert_group_call_participant,
|
|
heartbeat_group_call_participant,
|
|
mark_group_call_participant_left,
|
|
list_group_call_participants,
|
|
maybe_end_group_call_if_empty,
|
|
)
|
|
|
|
MAX_MESSAGE_LENGTH = CONSTANTS["MAX_MESSAGE_LENGTH"]
|
|
MAX_HISTORY_MESSAGES = CONSTANTS["MAX_HISTORY_MESSAGES"]
|
|
MAX_INCREMENTAL_MESSAGES = CONSTANTS["MAX_INCREMENTAL_MESSAGES"]
|
|
MAX_FEED_MESSAGES = CONSTANTS["MAX_FEED_MESSAGES"]
|
|
MAX_THREAD_REPLIES = CONSTANTS["MAX_THREAD_REPLIES"]
|
|
MAX_ATTACHMENT_BYTES = CONSTANTS["MAX_ATTACHMENT_BYTES"]
|
|
MAX_ATTACHMENT_NAME_LENGTH = CONSTANTS["MAX_ATTACHMENT_NAME_LENGTH"]
|
|
ALLOWED_ATTACHMENT_EXTENSIONS = CONSTANTS["ALLOWED_ATTACHMENT_EXTENSIONS"]
|
|
ALLOWED_ATTACHMENT_MIME_TYPES = CONSTANTS["ALLOWED_ATTACHMENT_MIME_TYPES"]
|
|
INVITE_TTL_HOURS = CONSTANTS["INVITE_TTL_HOURS"]
|
|
TEAM_INVITE_TTL_HOURS = CONSTANTS["TEAM_INVITE_TTL_HOURS"]
|
|
TYPING_TTL_SECONDS = CONSTANTS["TYPING_TTL_SECONDS"]
|
|
PRESENCE_TTL_SECONDS = CONSTANTS["PRESENCE_TTL_SECONDS"]
|
|
MAX_SEARCH_QUERY_LENGTH = CONSTANTS["MAX_SEARCH_QUERY_LENGTH"]
|
|
MAX_SEARCH_RESULTS = CONSTANTS["MAX_SEARCH_RESULTS"]
|
|
READ_RECEIPT_BATCH_LIMIT = CONSTANTS["READ_RECEIPT_BATCH_LIMIT"]
|
|
MAX_MESSAGE_EDIT_WINDOW_SECONDS = CONSTANTS["MAX_MESSAGE_EDIT_WINDOW_SECONDS"]
|
|
MAX_DISPLAY_NAME_LENGTH = CONSTANTS["MAX_DISPLAY_NAME_LENGTH"]
|
|
MAX_PROFILE_BIO_LENGTH = CONSTANTS["MAX_PROFILE_BIO_LENGTH"]
|
|
MAX_LOCATION_LENGTH = CONSTANTS["MAX_LOCATION_LENGTH"]
|
|
MAX_WEBSITE_LENGTH = CONSTANTS["MAX_WEBSITE_LENGTH"]
|
|
MAX_CUSTOM_CSS_LENGTH = CONSTANTS["MAX_CUSTOM_CSS_LENGTH"]
|
|
MAX_PROFILE_BADGE_LENGTH = CONSTANTS["MAX_PROFILE_BADGE_LENGTH"]
|
|
MAX_OWNED_TEAMS_PER_USER = CONSTANTS["MAX_OWNED_TEAMS_PER_USER"]
|
|
GROUP_CALL_PARTICIPANT_STALE_SECONDS = CONSTANTS["GROUP_CALL_PARTICIPANT_STALE_SECONDS"]
|
|
ROOM_VISIBILITIES = CONSTANTS["ROOM_VISIBILITIES"]
|
|
ROLES = CONSTANTS["ROLES"]
|
|
STATUSES = CONSTANTS["STATUSES"]
|
|
|
|
|
|
def private_access_key_valid(candidate: object) -> bool:
|
|
configured = (PRIVATE_ACCESS_KEY or "").strip()
|
|
if not configured:
|
|
return True
|
|
if not isinstance(candidate, str):
|
|
return False
|
|
return candidate.strip() == configured
|
|
|
|
|
|
def build_message_payload(row, user):
|
|
return richer_message_payload(
|
|
row,
|
|
user,
|
|
attachment_payload_legacy_from_message_select_row,
|
|
attachment_list_for_message,
|
|
)
|
|
|
|
|
|
def build_feed_payload(row, user):
|
|
return feed_message_payload(
|
|
row,
|
|
user,
|
|
attachment_payload_legacy_from_message_select_row,
|
|
attachment_list_for_message,
|
|
)
|
|
|
|
|
|
def build_local_stub_share_url(kind: str, object_id: int) -> str:
|
|
return f"/shared/{kind}/{object_id}"
|
|
|
|
|
|
def require_csrf():
|
|
if not csrf_valid():
|
|
return jsonify({"error": "Security token expired.", "code": "csrf"}), 403
|
|
return None
|
|
|
|
|
|
def parse_json_request():
|
|
try:
|
|
return safe_json(), None
|
|
except ValueError as error:
|
|
return None, (jsonify({"error": str(error)}), 400)
|
|
|
|
|
|
def require_dm_access(dm_id: int, user):
|
|
if not validate_dm_access(dm_id, user):
|
|
return None, (jsonify({"error": "DM not found."}), 404)
|
|
return dm_id, None
|
|
|
|
|
|
def require_room_access(
|
|
slug: str,
|
|
user,
|
|
login_message="Login required.",
|
|
forbidden_message="You do not have access to this room.",
|
|
):
|
|
room = get_room_by_slug(slug)
|
|
if not room:
|
|
return None, (jsonify({"error": "Room not found."}), 404)
|
|
|
|
if not can_access_room(room, user):
|
|
if not user:
|
|
return None, (jsonify({"error": login_message, "code": "login_required"}), 401)
|
|
return None, (jsonify({"error": forbidden_message, "code": "forbidden"}), 403)
|
|
|
|
return room, None
|
|
|
|
|
|
def require_room_write(slug: str, user):
|
|
room = get_room_by_slug(slug)
|
|
if not room:
|
|
return None, (jsonify({"error": "Room not found."}), 404)
|
|
|
|
if not can_write_room(room, user):
|
|
return None, (jsonify({"error": "You cannot post in this room.", "code": "forbidden"}), 403)
|
|
|
|
return room, None
|
|
|
|
|
|
def require_channel_access(team_slug: str, channel_slug: str, user):
|
|
channel = get_channel_by_slug(team_slug, channel_slug)
|
|
if not channel:
|
|
return None, (jsonify({"error": "Channel not found."}), 404)
|
|
|
|
if not can_access_channel(channel, user):
|
|
if not user:
|
|
return None, (jsonify({"error": "Login required.", "code": "login_required"}), 401)
|
|
return None, (jsonify({"error": "You do not have access to this channel.", "code": "forbidden"}), 403)
|
|
|
|
return channel, None
|
|
|
|
|
|
def require_channel_write(team_slug: str, channel_slug: str, user):
|
|
channel = get_channel_by_slug(team_slug, channel_slug)
|
|
if not channel:
|
|
return None, (jsonify({"error": "Channel not found."}), 404)
|
|
|
|
if not can_write_channel(channel, user):
|
|
return None, (jsonify({"error": "You cannot post in this channel.", "code": "forbidden"}), 403)
|
|
|
|
return channel, None
|
|
|
|
|
|
def user_can_access_message_scope(message_row, user) -> bool:
|
|
if not message_row:
|
|
return False
|
|
|
|
if message_row["direct_conversation_id"]:
|
|
return bool(user and validate_dm_access(message_row["direct_conversation_id"], user))
|
|
|
|
if message_row["channel_id"]:
|
|
channel = get_channel_by_id(message_row["channel_id"])
|
|
return bool(channel and can_access_channel(channel, user))
|
|
|
|
if message_row["room_id"]:
|
|
room = room_by_id(message_row["room_id"])
|
|
return bool(room and can_access_room(room, user))
|
|
|
|
return False
|
|
|
|
|
|
def resolve_message_attachment_inputs(data: dict, user_id: int):
|
|
attachment_ids, attachment_error = resolve_owned_attachment_ids(
|
|
data.get("attachment_ids", []),
|
|
user_id,
|
|
)
|
|
if attachment_error:
|
|
return None, None, attachment_error
|
|
|
|
if not attachment_ids and data.get("attachment_id") is not None:
|
|
single_attachment_id, single_attachment_error = resolve_owned_attachment(
|
|
data.get("attachment_id"),
|
|
user_id,
|
|
)
|
|
if single_attachment_error:
|
|
return None, None, single_attachment_error
|
|
attachment_ids = [single_attachment_id] if single_attachment_id else []
|
|
|
|
attachment_id = attachment_ids[0] if attachment_ids else None
|
|
return attachment_ids, attachment_id, None
|
|
|
|
|
|
def require_team_admin(team_slug: str, user):
|
|
team = get_team_by_slug(team_slug)
|
|
if not team:
|
|
return None, (jsonify({"error": "Team not found."}), 404)
|
|
|
|
if not is_team_admin(team["id"], user):
|
|
return None, (jsonify({"error": "Team admin access required.", "code": "forbidden"}), 403)
|
|
|
|
return team, None
|
|
|
|
|
|
def require_team_access(team_slug: str, user):
|
|
team = get_team_by_slug(team_slug)
|
|
if not team:
|
|
return None, (jsonify({"error": "Team not found."}), 404)
|
|
|
|
if not can_access_team(team, user):
|
|
if not user:
|
|
return None, (jsonify({"error": "Login required.", "code": "login_required"}), 401)
|
|
return None, (jsonify({"error": "You do not have access to this team.", "code": "forbidden"}), 403)
|
|
|
|
return team, None
|
|
|
|
|
|
def load_active_page(team_slug: str, page_slug: str):
|
|
with get_db() as db:
|
|
page = db.execute(
|
|
f"""
|
|
{team_page_select()}
|
|
WHERE t.slug = ?
|
|
AND p.slug = ?
|
|
AND p.is_active = 1
|
|
AND t.is_active = 1
|
|
""",
|
|
(team_slug, page_slug),
|
|
).fetchone()
|
|
return page
|
|
|
|
|
|
def can_read_page(page, user) -> bool:
|
|
return bool(
|
|
page
|
|
and (
|
|
page["visibility"] == "public"
|
|
or (
|
|
user
|
|
and (
|
|
user["role"] in {"moderator", "admin"}
|
|
or is_team_member(page["team_id"], user["id"])
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
def can_edit_page(page, user) -> bool:
|
|
return bool(page and user and is_team_admin(page["team_id"], user))
|
|
|
|
|
|
def can_access_group_call_session(session_row, user: dict) -> bool:
|
|
if not session_row or not user:
|
|
return False
|
|
|
|
if session_row["scope_kind"] == "room":
|
|
room = room_by_id(session_row["scope_id"])
|
|
return bool(room and can_access_room(room, user))
|
|
|
|
if session_row["scope_kind"] == "channel":
|
|
channel = get_channel_by_id(session_row["scope_id"])
|
|
return bool(channel and can_access_channel(channel, user))
|
|
|
|
return False
|
|
|
|
|
|
def can_manage_group_call_session(session_row, user: dict) -> bool:
|
|
if not session_row or not user:
|
|
return False
|
|
|
|
if user["role"] in {"moderator", "admin"}:
|
|
return True
|
|
|
|
if int(session_row["created_by_user_id"]) == int(user["id"]):
|
|
return True
|
|
|
|
if session_row["scope_kind"] == "room":
|
|
room = room_by_id(session_row["scope_id"])
|
|
return bool(room and can_manage_room(room, user))
|
|
|
|
if session_row["scope_kind"] == "channel":
|
|
channel = get_channel_by_id(session_row["scope_id"])
|
|
return bool(channel and can_edit_team(channel["team_id"], user))
|
|
|
|
return False
|
|
|
|
|
|
def group_call_scope_payload(session_row, user):
|
|
payload = {
|
|
"scope_kind": session_row["scope_kind"],
|
|
"scope_id": session_row["scope_id"],
|
|
"scope_slug": session_row["scope_slug"],
|
|
}
|
|
|
|
if session_row["scope_kind"] == "room":
|
|
room = room_by_id(session_row["scope_id"])
|
|
if room:
|
|
payload["room"] = room_payload(room, user)
|
|
payload["scope_label"] = room["name"]
|
|
elif session_row["scope_kind"] == "channel":
|
|
channel = get_channel_by_id(session_row["scope_id"])
|
|
if channel:
|
|
payload["channel"] = channel_payload(channel, user)
|
|
payload["team"] = {
|
|
"slug": channel["team_slug"],
|
|
"name": channel["team_name"],
|
|
}
|
|
payload["scope_label"] = f"# {channel['name']}"
|
|
|
|
return payload
|
|
|
|
|
|
def group_call_permissions_payload(session_row, user):
|
|
can_manage = can_manage_group_call_session(session_row, user)
|
|
can_access = can_access_group_call_session(session_row, user)
|
|
return {
|
|
"can_manage": can_manage,
|
|
"can_end": can_manage,
|
|
"can_invite": can_access,
|
|
}
|