From 7d3333bbff7c3589fee599b9728586c9281472f6 Mon Sep 17 00:00:00 2001 From: Binbim_ProMax Date: Sat, 4 Jul 2026 07:15:41 +0800 Subject: [PATCH] feat: migrate to LiveKit, remove Agora/payment, add domain certs for Caddy --- Cargo.lock | 21 +- Cargo.toml | 1 - DEPLOY_LIVEKIT.md | 156 ++++++++++++++ README.md | 10 +- config/config.toml | 5 +- crates/agora-token/Cargo.toml | 12 -- crates/agora-token/src/lib.rs | 98 --------- deploy/Caddyfile | 28 +++ deploy/Dockerfile | 26 +++ deploy/docker-compose.yaml | 56 +++++ deploy/generate-keys.sh | 17 ++ deploy/livekit.yaml | 26 +++ src/api/admin_agora.rs | 155 -------------- src/api/admin_livekit.rs | 382 ++++++++++++++++++++++++++++++++++ src/api/admin_system.rs | 59 ++++++ src/api/group.rs | 56 +---- src/api/license.rs | 148 ++++--------- src/api/message.rs | 19 ++ src/api/mod.rs | 13 +- src/api/tags.rs | 4 +- src/api/user.rs | 14 ++ src/create_user.rs | 17 -- src/license.rs | 106 +++------- src/server.rs | 4 +- src/state.rs | 5 + 25 files changed, 875 insertions(+), 563 deletions(-) create mode 100644 DEPLOY_LIVEKIT.md delete mode 100644 crates/agora-token/Cargo.toml delete mode 100644 crates/agora-token/src/lib.rs create mode 100644 deploy/Caddyfile create mode 100644 deploy/Dockerfile create mode 100644 deploy/docker-compose.yaml create mode 100644 deploy/generate-keys.sh create mode 100644 deploy/livekit.yaml delete mode 100644 src/api/admin_agora.rs create mode 100644 src/api/admin_livekit.rs diff --git a/Cargo.lock b/Cargo.lock index d76e1eb..1719c0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "adler" @@ -43,18 +43,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "agora-token" -version = "0.1.0" -dependencies = [ - "base64 0.13.1", - "checksum", - "fastrand", - "hmac 0.11.0", - "sha2 0.9.9", - "thiserror", -] - [[package]] name = "ahash" version = "0.7.6" @@ -373,12 +361,6 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" -[[package]] -name = "checksum" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5c24f6a463e9973db3df3c2cc276f689f5baf289c87a693dc859e004d3eb45f" - [[package]] name = "chrono" version = "0.4.23" @@ -4672,7 +4654,6 @@ checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" name = "vocechat-server" version = "0.3.3" dependencies = [ - "agora-token", "anyhow", "async-stream", "base64 0.13.1", diff --git a/Cargo.toml b/Cargo.toml index 33f5c13..e71e697 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,6 @@ rc-token = { path = "./crates/token" } rc-fcm = { path = "./crates/fcm" } rc-magic-link = { path = "./crates/magic-link" } vc-license = { path = "./crates/vc-license" } -agora-token = { path = "./crates/agora-token" } open-graph = { path = "./crates/open-graph", features = ["poem_openapi"] } poem = { version = "1.3.51", features = [ diff --git a/DEPLOY_LIVEKIT.md b/DEPLOY_LIVEKIT.md new file mode 100644 index 0000000..5c6cc1c --- /dev/null +++ b/DEPLOY_LIVEKIT.md @@ -0,0 +1,156 @@ +# ColdBreeze Chat + LiveKit 部署指南 + +## 架构 + +``` +浏览器 ──HTTPS──> VoceChat (3000) ──> 聊天/管理 + └──WS/HTTP──> LiveKit (7880) ──> 音视频信令 + └──UDP──────> LiveKit (50000-50100) ──> WebRTC 媒体流 +``` + +VoceChat 负责聊天和管理，LiveKit 负责音视频通话和屏幕共享。两者通过 Docker Compose 一键部署。 + +## 前置条件 + +- Rust 工具链（用于编译服务端） +- Docker + Docker Compose +- 端口开放：3000、7880、7881、50000-50100/udp + +## 快速启动 + +### 0. 编译服务端二进制（或使用 Docker 多阶段构建） + +NAS 上无需安装 Rust。`docker compose build` 会在 Docker 内自动编译。 + +如需本地编译： + +```bash +cd ColdBreeze-chat-server-rust +cargo build --release +cp target/release/vocechat-server . +``` + +### 0.5 构建 Web 客户端 + +```bash +cd ColdBreeze-chat-web +pnpm install +REACT_APP_BUILD_TIME=$(date +%s) GENERATE_SOURCEMAP=false node scripts/build.js +``` + +将 `build/` 目录内容上传到 NAS 的 `/vol1/docker/coldbreeze/web-client/`。 + +### 1. 生成 API Key / Secret + +```bash +cd deploy +bash generate-keys.sh +``` + +将输出的 key 和 secret 填入 `livekit.yaml` 的 `keys` 部分： + +```yaml +keys: + <你的API_KEY>: <你的API_SECRET> +``` + +### 2. 启动服务 + +```bash +cd deploy +docker compose build +docker compose up -d +``` + +### 3. 初始化 VoceChat + +浏览器访问 `http://你的NAS_IP:3009`，完成管理员注册。 + +### 4. 配置 LiveKit + +进入 VoceChat 管理后台 → 设置 → video（LiveKit）： + +| 字段 | 值 | +|------|-----| +| Enable | ✅ | +| LiveKit URL | `ws://你的NAS_IP:7880` | +| API Key | 同 livekit.yaml 中的 key | +| API Secret | 同 livekit.yaml 中的 secret | + +点击保存。 + +### 5. 测试 + +进入任意频道，点击语音/视频按钮，确认能正常加入 LiveKit 房间。 + +## 端口说明 + +| 端口 | 协议 | 用途 | +|------|------|------| +| 3009 | TCP | VoceChat Web 客户端 + API | +| 7880 | TCP | LiveKit 信令（WebSocket） | +| 7881 | TCP | WebRTC over TCP fallback | +| 50000-50100 | UDP | WebRTC 媒体流（音视频数据） | + +## IPv6 / 动态公网 + +如果需要公网访问： + +1. 确保路由器开启了 IPv6，并给 NAS 分配了公网 IPv6 地址 +2. DNS AAAA 记录指向 NAS 的公网 IPv6 +3. 防火墙放行上述端口 +4. VoceChat 管理后台 → 设置 → 域名，填入你的域名 +5. LiveKit URL 改为 `ws://你的域名:7880` + +如果 LiveKit 需要从公网可达，确保 UDP 50000-50100 端口未被运营商封锁。 + +## 常用命令 + +```bash +# 查看日志 +docker compose logs -f livekit-server +docker compose logs -f vocechat-server + +# 重启 +docker compose restart + +# 停止 +docker compose down + +# 重新生成 key 后更新配置 +# 1. 修改 livekit.yaml 中的 keys +# 2. 重启 LiveKit +docker compose restart livekit-server +# 3. 在 VoceChat 管理后台更新对应的 key/secret +``` + +## LiveKit 配置参考 + +`livekit.yaml` 完整配置项： + +```yaml +port: 7880 # 信令端口 + +rtc: + port_range_start: 50000 # UDP 媒体端口范围 + port_range_end: 50100 + tcp_port: 7881 # TCP fallback + use_external_ip: false # 内网部署保持 false + +keys: + your_api_key: your_api_secret + +logging: + level: info # debug / info / warn / error + +room: + auto_create: true # 自动创建房间 + empty_timeout: 300 # 空房间 5 分钟后关闭 + max_participants: 0 # 不限制 +``` + +## 注意事项 + +- LiveKit 走内网 HTTP，不加密。如果浏览器通过 HTTPS 访问 VoceChat，可能需要在浏览器设置中允许混合内容（或使用内网 HTTP 直接访问）。 +- UDP 端口 50000-50100 是 WebRTC 媒体流必需的，如果音视频不通，优先检查防火墙和路由器是否放行了这些端口。 +- NAS 重启后 Docker 会自动拉起服务（`restart: unless-stopped`）。 diff --git a/README.md b/README.md index d43b591..b8cb200 100644 --- a/README.md +++ b/README.md @@ -44,11 +44,9 @@ We also found bot super interesting and personal bot services may replace platfo If you have a proposal on how multiple servers could get interconnected, e.g., should we support Matrix protocal, or other protocal, feel free to share your thougts in the discussions, or directly discuss it with at our chat at https://voce.chat. -# Free for personal use, require license for non-personal use. +# ColdBreeze private deployment -`VoceChat` is an open-source commercial software. The VoceChat server is the smallest, stablest and most efficient independent chat server on today's market, which is good for integration to your own app. VoceChat official image is free for personal use, which we define as equal or less than 20 registered users, if you want to integrate VoceChat to your own app/site for a larger user base, you have to purchase a license. The license is 49$/version. - -Our team also provide customization service. We also provide resale license for NAS and cloud providers who want to collaborate with us. +This ColdBreeze build is intended for fully self-hosted deployments. The external purchase flow and user-count upgrade requirement are removed, and the server uses a local unrestricted license state at runtime. ### Project composition: @@ -62,8 +60,8 @@ Our team also provide customization service. We also provide resale license for ### Feature list & roadmap -- [x] Voice (now support VoceSpace private hosting solution, also agora) -- [x] Video (now support VoceSpace private hosting solution, also agora) +- [x] Voice (self-hosted LiveKit) +- [x] Video (self-hosted LiveKit) - [x] DM & Group Chating - [x] Reply, @ to mention a person - [x] Images and large files transmission diff --git a/config/config.toml b/config/config.toml index 09b2949..5ad7a56 100644 --- a/config/config.toml +++ b/config/config.toml @@ -1,5 +1,6 @@ # webclient_url = "https://github.com/Privoce/vocechat-web/releases/latest/download" -webclient_url = "https://s.voce.chat/web_client/v0.3.x" +# webclient_url = "https://s.voce.chat/web_client/v0.3.x" +webclient_url = "" [system] data_dir = "./data" @@ -10,7 +11,7 @@ refresh_token_expiry_seconds = 604800 [network] bind = "0.0.0.0:3000" # domain = "chat.privoce.com" -frontend_url = "http://1.2.3.4:4000" +frontend_url = "http://192.168.1.122:3009" # [network.tls] # type = "self_signed" diff --git a/crates/agora-token/Cargo.toml b/crates/agora-token/Cargo.toml deleted file mode 100644 index 1ec29c4..0000000 --- a/crates/agora-token/Cargo.toml +++ /dev/null @@ -1,12 +0,0 @@ -[package] -name = "agora-token" -version = "0.1.0" -edition = "2021" - -[dependencies] -base64 = "0.13.0" -checksum = "0.2.1" -fastrand = "1.6.0" -hmac = "0.11.0" -sha2 = "0.9" -thiserror = "1.0.30" diff --git a/crates/agora-token/src/lib.rs b/crates/agora-token/src/lib.rs deleted file mode 100644 index 3284d1e..0000000 --- a/crates/agora-token/src/lib.rs +++ /dev/null @@ -1,98 +0,0 @@ -use std::time::{SystemTime, UNIX_EPOCH}; - -use hmac::{Mac, NewMac}; -use sha2::Sha256; - -const VERSION: &str = "006"; - -#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd)] -enum Privileges { - JoinChannel = 1, - PublishAudioStream = 2, - PublishVideoStream = 3, - PublishDataStream = 4, -} - -fn create_access_token( - app_id: &str, - app_certificate: &str, - channel_name: &str, - uid: &str, - privileges: &[Privileges], -) -> String { - let ts = (SystemTime::now().duration_since(UNIX_EPOCH)) - .unwrap() - .as_secs() as u32 - + 24 * 3600; - let salt = fastrand::u32(1..99999999); - let mut buf_m = Vec::new(); - - buf_m.extend_from_slice(&salt.to_le_bytes()); - buf_m.extend_from_slice(&ts.to_le_bytes()); - - buf_m.extend_from_slice(&(privileges.len() as u16).to_le_bytes()); - let mut privileges = privileges.to_vec(); - privileges.sort(); - - for k in privileges { - buf_m.extend_from_slice(&(k as u16).to_le_bytes()); - buf_m.extend_from_slice(&ts.to_le_bytes()); - } - - let mut buf_val = Vec::new(); - buf_val.extend(format!("{}{}{}", app_id, channel_name, uid).as_bytes()); - buf_val.extend(&buf_m); - - let buf_sig = { - let mut mac = hmac::Hmac::::new_from_slice(app_certificate.as_bytes()) - .expect("valid app certificate"); - mac.update(&buf_val); - mac.finalize().into_bytes() - }; - - let crc_channel_name = crc32(channel_name.as_bytes()); - let crc_uid = crc32(uid.as_bytes()); - - let mut buf_content = Vec::new(); - pack_bytes(&mut buf_content, &buf_sig); - buf_content.extend_from_slice(&crc_channel_name.to_le_bytes()); - buf_content.extend_from_slice(&crc_uid.to_le_bytes()); - pack_bytes(&mut buf_content, &buf_m); - - format!("{}{}{}", VERSION, app_id, base64::encode(buf_content)) -} - -fn crc32(data: &[u8]) -> u32 { - let mut hasher = checksum::crc32::Crc32::new(); - hasher.checksum(data) -} - -fn pack_bytes(data: &mut Vec, s: &[u8]) { - data.extend_from_slice(&(s.len() as u16).to_le_bytes()); - data.extend_from_slice(s); -} - -pub fn create_rtc_token( - app_id: &str, - app_certificate: &str, - channel_name: &str, - uid: u32, -) -> String { - let uid = if uid == 0 { - String::new() - } else { - uid.to_string() - }; - create_access_token( - app_id, - app_certificate, - channel_name, - &uid, - &[ - Privileges::JoinChannel, - Privileges::PublishAudioStream, - Privileges::PublishVideoStream, - Privileges::PublishDataStream, - ], - ) -} diff --git a/deploy/Caddyfile b/deploy/Caddyfile new file mode 100644 index 0000000..e64e97e --- /dev/null +++ b/deploy/Caddyfile @@ -0,0 +1,28 @@ +{ + # Disable automatic HTTPS since we provide our own certs + auto_https off +} + +# VoceChat - vocechat.binbim.locker +vocechat.binbim.locker { + tls /certs/vocechat/cert.pem /certs/vocechat/key.pem + + # VoceChat API + Web client + handle { + reverse_proxy vocechat-server:3000 + } + + # LiveKit signaling via /rtc path + handle /rtc/* { + reverse_proxy livekit-server:7880 + } +} + +# LiveKit - livekit.binbim.locker (for direct WSS access if needed) +livekit.binbim.locker { + tls /certs/livekit/cert.pem /certs/livekit/key.pem + + handle { + reverse_proxy livekit-server:7880 + } +} \ No newline at end of file diff --git a/deploy/Dockerfile b/deploy/Dockerfile new file mode 100644 index 0000000..0c6f101 --- /dev/null +++ b/deploy/Dockerfile @@ -0,0 +1,26 @@ +# Stage 1: Build the Rust server binary +FROM rust:1.82-bookworm AS builder + +WORKDIR /src +COPY Cargo.toml Cargo.lock ./ +COPY src ./src +COPY crates ./crates +COPY migrations ./migrations +COPY cert ./cert + +RUN cargo build --release && \ + strip target/release/vocechat-server + +# Stage 2: Minimal runtime image +FROM debian:bookworm-slim + +RUN apt-get update && \ + apt-get install -y --no-install-recommends ca-certificates && \ + rm -rf /var/lib/apt/lists/* + +COPY --from=builder /src/target/release/vocechat-server /home/vocechat-server/vocechat-server +COPY config /home/vocechat-server/config + +EXPOSE 3000 +WORKDIR /home/vocechat-server +CMD ["./vocechat-server"] diff --git a/deploy/docker-compose.yaml b/deploy/docker-compose.yaml new file mode 100644 index 0000000..c99dec9 --- /dev/null +++ b/deploy/docker-compose.yaml @@ -0,0 +1,56 @@ +services: + vocechat-server: + # Multi-stage build: compiles Rust binary from source inside Docker. + # No local Rust toolchain needed. First build takes a few minutes. + build: + context: .. + dockerfile: deploy/Dockerfile + container_name: coldbreeze-server + restart: unless-stopped + volumes: + - vocechat_data:/home/vocechat-server/data + - /vol1/docker/coldbreeze/web-client:/home/vocechat-server/data/wwwroot:ro + expose: + - "3000" + depends_on: + - livekit-server + + livekit-server: + image: livekit/livekit-server:v1.13.3 + container_name: coldbreeze-livekit + restart: unless-stopped + command: + - --bind + - 0.0.0.0 + - --config + - /etc/livekit.yaml + volumes: + - ./livekit.yaml:/etc/livekit.yaml:ro + ports: + # WebRTC media (UDP) - must be host-mapped for browser ICE + - "50000-50100:50000-50100/udp" + expose: + - "7880" + - "7881" + + caddy: + image: caddy:2-alpine + container_name: coldbreeze-caddy + restart: unless-stopped + volumes: + - ./Caddyfile:/etc/caddy/Caddyfile:ro + - /vol1/docker/coldbreeze/certs/vocechat:/certs/vocechat:ro + - /vol1/docker/coldbreeze/certs/livekit:/certs/livekit:ro + - caddy_data:/data + - caddy_config:/config + ports: + # HTTPS (443) - the only public entry point + - "443:443" + depends_on: + - vocechat-server + - livekit-server + +volumes: + vocechat_data: + caddy_data: + caddy_config: \ No newline at end of file diff --git a/deploy/generate-keys.sh b/deploy/generate-keys.sh new file mode 100644 index 0000000..414fd4d --- /dev/null +++ b/deploy/generate-keys.sh @@ -0,0 +1,17 @@ +#!/bin/bash +# Generate a LiveKit API key/secret pair. +# Copy the output into: +# 1. deploy/livekit.yaml (keys section) +# 2. VoceChat admin -> Settings -> video (LiveKit) config page + +set -e + +echo "Generating LiveKit API key and secret..." +echo "" + +docker run --rm livekit/livekit-server generate-keys + +echo "" +echo "Copy the API Key and Secret into:" +echo " 1. deploy/livekit.yaml -> keys section" +echo " 2. VoceChat admin panel -> Settings -> video (LiveKit)" diff --git a/deploy/livekit.yaml b/deploy/livekit.yaml new file mode 100644 index 0000000..cbbb1e9 --- /dev/null +++ b/deploy/livekit.yaml @@ -0,0 +1,26 @@ +# LiveKit Server Configuration +# Docs: https://docs.livekit.io/oss/deployment/configuration + +port: 7880 + +rtc: + port_range_start: 50000 + port_range_end: 50100 + tcp_port: 7881 + use_external_ip: false + +# API key/secret pairs for JWT authentication. +# Generate your own: docker run --rm livekit/livekit-server generate-keys +# Then replace the key and secret below, and enter the same values in +# VoceChat admin -> Settings -> video (LiveKit). +keys: + coldbreeze: CHANGE_ME_TO_YOUR_GENERATED_SECRET + +logging: + level: info + json: false + +room: + auto_create: true + empty_timeout: 300 + max_participants: 0 diff --git a/src/api/admin_agora.rs b/src/api/admin_agora.rs deleted file mode 100644 index b3816e6..0000000 --- a/src/api/admin_agora.rs +++ /dev/null @@ -1,155 +0,0 @@ -use poem::{ - error::{InternalServerError, ServiceUnavailable}, - http::StatusCode, - web::Data, - Error, Result, -}; -use poem_openapi::{param::Query, payload::Json, Object, OpenApi}; -use serde::{Deserialize, Serialize}; - -use crate::{ - api::{tags::ApiTags, token::Token, DateTime}, - config::Config, - state::{DynamicConfig, DynamicConfigEntry}, - State, -}; - -pub struct ApiAdminAgora; - -#[derive(Debug, Object, Serialize, Deserialize, Default)] -pub struct AgoraConfig { - #[oai(default = "default_agora_url")] - pub url: String, - pub project_id: String, - pub app_id: String, - pub app_certificate: String, - pub rtm_key: String, - pub rtm_secret: String, -} - -fn default_agora_url() -> String { - "https://api.agora.io".to_string() -} - -impl DynamicConfig for AgoraConfig { - type Instance = Self; - - fn name() -> &'static str { - "agora" - } - - fn create_instance(self, _config: &Config) -> Self::Instance { - self - } -} - -/// Agora config -#[derive(Debug, Object)] -pub struct AgoraConfigObject { - enabled: bool, - #[oai(flatten)] - config: AgoraConfig, -} - -/// Agora usage response -#[derive(Deserialize, Object)] -struct AgoraUsagesResponse { - usages: Vec, -} - -/// Agora usage item -#[derive(Deserialize, Object)] -struct AgoraUsageItem { - date: DateTime, - usage: AgoraUsage, -} - -/// Agora usage -#[derive(Deserialize, Object)] -#[serde(rename_all = "camelCase")] -struct AgoraUsage { - duration_audio_all: i64, - duration_video_hd: i64, - duration_video_hdp: i64, -} - -#[OpenApi(prefix_path = "/admin/agora", tag = "ApiTags::AdminAgora")] -impl ApiAdminAgora { - /// Set Agora config - #[oai(path = "/config", method = "post")] - async fn set_config( - &self, - state: Data<&State>, - token: Token, - config: Json, - ) -> Result<()> { - if !token.is_admin { - return Err(Error::from_status(StatusCode::FORBIDDEN)); - } - state - .set_dynamic_config(DynamicConfigEntry { - enabled: config.0.enabled, - config: config.0.config, - }) - .await?; - - Ok(()) - } - - /// Get Agora config - #[oai(path = "/config", method = "get")] - async fn get_config( - &self, - state: Data<&State>, - _token: Token, - ) -> Result> { - // if !token.is_admin { - // return Err(Error::from_status(StatusCode::FORBIDDEN)); - // } - let entry = state.load_dynamic_config::().await?; - Ok(Json(AgoraConfigObject { - enabled: entry.enabled, - config: entry.config, - })) - } - - /// Get Agora usage - #[oai(path = "/usages", method = "get")] - async fn usage( - &self, - state: Data<&State>, - token: Token, - /// Start date(YYYY-MM-DD) - from_date: Query, - /// End date(YYYY-MM-DD) - to_date: Query, - ) -> Result> { - if !token.is_admin { - return Err(Error::from_status(StatusCode::FORBIDDEN)); - } - - let agora = state - .get_dynamic_config_instance::() - .await - .ok_or_else(|| Error::from_status(StatusCode::SERVICE_UNAVAILABLE))?; - let resp = reqwest::Client::new() - .get(format!("{}/dev/v3/usage", agora.url)) - .query(&[ - ("project_id", agora.project_id.as_str()), - ("from_date", from_date.as_str()), - ("to_date", to_date.as_str()), - ("business", "default"), - ]) - .header("accept", "application/json") - .basic_auth(&agora.rtm_key, Some(&agora.rtm_secret)) - .send() - .await - .map_err(ServiceUnavailable)? - .error_for_status() - .map_err(ServiceUnavailable)? - .json::() - .await - .map_err(InternalServerError)?; - Ok(Json(resp)) - } -} diff --git a/src/api/admin_livekit.rs b/src/api/admin_livekit.rs new file mode 100644 index 0000000..c42d683 --- /dev/null +++ b/src/api/admin_livekit.rs @@ -0,0 +1,382 @@ +use std::{collections::BTreeSet, sync::Arc}; + +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use poem::{ + error::InternalServerError, + http::StatusCode, + web::Data, + Error, Result, +}; +use poem_openapi::{payload::Json, Object, OpenApi}; +use serde::{Deserialize, Serialize}; + +use crate::{ + api::{tags::ApiTags, token::Token, RtcCallAction, RtcCallMessage}, + config::Config, + state::{BroadcastEvent, DynamicConfig, DynamicConfigEntry}, + State, +}; + +const LIVEKIT_TOKEN_EXPIRED_IN: i64 = 24 * 3600; + +pub struct ApiAdminLivekit; + +#[derive(Debug, Object, Serialize, Deserialize, Default, Clone)] +pub struct LivekitConfig { + pub url: String, + pub api_key: String, + pub api_secret: String, +} + +impl DynamicConfig for LivekitConfig { + type Instance = Self; + + fn name() -> &'static str { + "livekit" + } + + fn create_instance(self, _config: &Config) -> Self::Instance { + self + } +} + +#[derive(Debug, Object)] +pub struct LivekitConfigObject { + enabled: bool, + #[oai(flatten)] + config: LivekitConfig, +} + +#[derive(Debug, Object)] +struct LivekitTokenRequest { + gid: Option, + uid: Option, +} + +#[derive(Debug, Object)] +struct LivekitTokenResponse { + token: String, + url: String, + room: String, + identity: String, + expired_in: i64, +} + +#[derive(Debug, Object)] +struct LivekitCallRequest { + uid: i64, + action: RtcCallAction, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct LivekitVideoGrant { + room_join: bool, + room: String, + can_publish: bool, + can_subscribe: bool, + can_publish_data: bool, +} + +#[derive(Debug, Serialize)] +struct LivekitClaims { + exp: i64, + nbf: i64, + iss: String, + sub: String, + name: String, + video: LivekitVideoGrant, +} + +fn dm_room_name(uid1: i64, uid2: i64) -> String { + let (min_uid, max_uid) = if uid1 <= uid2 { + (uid1, uid2) + } else { + (uid2, uid1) + }; + format!("coldbreeze:dm:{}:{}", min_uid, max_uid) +} + +fn group_room_name(gid: i64) -> String { + format!("coldbreeze:channel:{}", gid) +} + +fn normalize_livekit_url(url: &str) -> String { + url.trim().trim_end_matches('/').to_string() +} + +fn create_livekit_token( + config: &LivekitConfig, + room: String, + identity: String, + name: String, +) -> jsonwebtoken::errors::Result { + let now = chrono::Utc::now().timestamp(); + let claims = LivekitClaims { + exp: now + LIVEKIT_TOKEN_EXPIRED_IN, + nbf: now - 10, + iss: config.api_key.clone(), + sub: identity, + name, + video: LivekitVideoGrant { + room_join: true, + room, + can_publish: true, + can_subscribe: true, + can_publish_data: true, + }, + }; + encode( + &Header::new(Algorithm::HS256), + &claims, + &EncodingKey::from_secret(config.api_secret.as_bytes()), + ) +} + +#[OpenApi(prefix_path = "/admin/livekit", tag = "ApiTags::AdminLivekit")] +impl ApiAdminLivekit { + /// Set LiveKit config + #[oai(path = "/config", method = "post")] + async fn set_config( + &self, + state: Data<&State>, + token: Token, + config: Json, + ) -> Result<()> { + if !token.is_admin { + return Err(Error::from_status(StatusCode::FORBIDDEN)); + } + + let mut livekit_config = config.0.config; + livekit_config.url = normalize_livekit_url(&livekit_config.url); + + state + .set_dynamic_config(DynamicConfigEntry { + enabled: config.0.enabled, + config: livekit_config, + }) + .await?; + + Ok(()) + } + + /// Get LiveKit config + #[oai(path = "/config", method = "get")] + async fn get_config( + &self, + state: Data<&State>, + token: Token, + ) -> Result> { + if !token.is_admin { + return Err(Error::from_status(StatusCode::FORBIDDEN)); + } + + let entry = state.load_dynamic_config::().await?; + Ok(Json(LivekitConfigObject { + enabled: entry.enabled, + config: entry.config, + })) + } + + /// Check whether LiveKit is enabled + #[oai(path = "/enabled", method = "get")] + async fn enabled(&self, state: Data<&State>, _token: Token) -> Result> { + Ok(Json( + state + .get_dynamic_config_instance::() + .await + .is_some(), + )) + } + + /// Generate a LiveKit room token + #[oai(path = "/token", method = "post")] + async fn generate_token( + &self, + state: Data<&State>, + token: Token, + req: Json, + ) -> Result> { + let config = state + .get_dynamic_config_instance::() + .await + .ok_or_else(|| Error::from_status(StatusCode::SERVICE_UNAVAILABLE))?; + + if config.url.trim().is_empty() + || config.api_key.trim().is_empty() + || config.api_secret.trim().is_empty() + { + return Err(Error::from_status(StatusCode::SERVICE_UNAVAILABLE)); + } + + let cache = state.cache.read().await; + let current_user = cache + .users + .get(&token.uid) + .ok_or_else(|| Error::from_status(StatusCode::UNAUTHORIZED))?; + let room = match (req.gid, req.uid) { + (Some(gid), None) => { + let group = cache + .groups + .get(&gid) + .ok_or_else(|| Error::from_status(StatusCode::NOT_FOUND))?; + if !group.contains_user(token.uid) { + return Err(Error::from_status(StatusCode::FORBIDDEN)); + } + group_room_name(gid) + } + (None, Some(peer_uid)) => { + if !cache.users.contains_key(&peer_uid) { + return Err(Error::from_status(StatusCode::NOT_FOUND)); + } + dm_room_name(token.uid, peer_uid) + } + _ => return Err(Error::from_status(StatusCode::BAD_REQUEST)), + }; + + let identity = token.uid.to_string(); + let name = current_user.name.clone(); + let jwt = create_livekit_token(&config, room.clone(), identity.clone(), name) + .map_err(InternalServerError)?; + + Ok(Json(LivekitTokenResponse { + token: jwt, + url: normalize_livekit_url(&config.url), + room, + identity, + expired_in: LIVEKIT_TOKEN_EXPIRED_IN, + })) + } + + /// Send a LiveKit direct-message call signal + #[oai(path = "/call", method = "post")] + async fn signal_call( + &self, + state: Data<&State>, + token: Token, + req: Json, + ) -> Result<()> { + state + .get_dynamic_config_instance::() + .await + .ok_or_else(|| Error::from_status(StatusCode::SERVICE_UNAVAILABLE))?; + + if req.uid == token.uid { + return Err(Error::from_status(StatusCode::BAD_REQUEST)); + } + + let cache = state.cache.read().await; + if !cache.users.contains_key(&token.uid) { + return Err(Error::from_status(StatusCode::UNAUTHORIZED)); + } + if !cache.users.contains_key(&req.uid) { + return Err(Error::from_status(StatusCode::NOT_FOUND)); + } + drop(cache); + + let room = dm_room_name(token.uid, req.uid); + let _ = state.event_sender.send(Arc::new(BroadcastEvent::RtcCall { + targets: BTreeSet::from([req.uid]), + message: RtcCallMessage { + from_uid: token.uid, + to_uid: req.uid, + room, + action: req.action, + }, + })); + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use futures_util::StreamExt; + use serde_json::json; + + use crate::{api::LivekitConfig, state::DynamicConfigEntry, test_harness::TestServer}; + + #[tokio::test] + async fn test_livekit_token_for_group() { + let server = TestServer::new().await; + let state = server.state(); + state + .set_dynamic_config(DynamicConfigEntry { + enabled: true, + config: LivekitConfig { + url: "https://rtc.example.com/".to_string(), + api_key: "devkey".to_string(), + api_secret: "devsecret".to_string(), + }, + }) + .await + .unwrap(); + + let admin_token = server.login_admin().await; + let resp = server + .post("/api/group") + .header("X-API-Key", &admin_token) + .body_json(&json!({ + "name": "test", + "members": [], + })) + .send() + .await; + resp.assert_status_is_ok(); + let gid = resp.json().await.value().object().get("gid").i64(); + + let resp = server + .post("/api/admin/livekit/token") + .header("X-API-Key", &admin_token) + .body_json(&json!({ "gid": gid })) + .send() + .await; + resp.assert_status_is_ok(); + let json = resp.json().await; + let obj = json.value().object(); + obj.get("url").assert_string("https://rtc.example.com"); + obj.get("room") + .assert_string(&format!("coldbreeze:channel:{}", gid)); + obj.get("identity").assert_string("1"); + obj.get("token").assert_not_null(); + } + + #[tokio::test] + async fn test_livekit_call_signal_for_dm() { + let server = TestServer::new().await; + let state = server.state(); + state + .set_dynamic_config(DynamicConfigEntry { + enabled: true, + config: LivekitConfig { + url: "https://rtc.example.com/".to_string(), + api_key: "devkey".to_string(), + api_secret: "devsecret".to_string(), + }, + }) + .await + .unwrap(); + + let admin_token = server.login_admin().await; + let uid = server.create_user(&admin_token, "user1@voce.chat").await; + let user_token = server.login("user1@voce.chat").await; + let mut events = server.subscribe_events(&user_token, Some(&["rtc_call"])).await; + + let resp = server + .post("/api/admin/livekit/call") + .header("X-API-Key", &admin_token) + .body_json(&json!({ "uid": uid, "action": "invite" })) + .send() + .await; + resp.assert_status_is_ok(); + + let msg = events.next().await.unwrap(); + let obj = msg.value().object(); + obj.get("type").assert_string("rtc_call"); + obj.get("from_uid").assert_i64(1); + obj.get("to_uid").assert_i64(uid); + obj.get("room") + .assert_string(&format!("coldbreeze:dm:1:{}", uid)); + obj.get("action").assert_string("invite"); + } +} diff --git a/src/api/admin_system.rs b/src/api/admin_system.rs index 150a37c..4d362e9 100644 --- a/src/api/admin_system.rs +++ b/src/api/admin_system.rs @@ -16,6 +16,36 @@ use crate::{ State, }; +/// Server-level common settings +#[derive(Debug, Object, Serialize, Deserialize, Default, Clone)] +pub struct SystemCommon { + pub show_user_online_status: Option, + pub webclient_auto_update: Option, + pub contact_verification_enable: Option, + pub chat_layout_mode: Option, + pub max_file_expiry_mode: Option, + pub only_admin_can_create_group: Option, + pub who_can_invite_users: Option, + pub ext_setting: Option, + pub msg_smtp_notify_enable: Option, + pub msg_smtp_notify_delay_seconds: Option, + pub dm_enable: Option, + pub add_friend_enable: Option, + pub search_user_enable: Option, +} + +impl DynamicConfig for SystemCommon { + type Instance = Self; + + fn name() -> &'static str { + "system_common" + } + + fn create_instance(self, _config: &Config) -> Self::Instance { + self + } +} + /// Server metrics #[derive(Debug, Object)] pub struct Metrics { @@ -163,6 +193,35 @@ impl ApiAdminSystem { })) } + /// Get system common settings + #[oai(path = "/common", method = "get")] + async fn get_system_common(&self, state: Data<&State>) -> Json { + match state.load_dynamic_config::().await { + Ok(entry) => Json(entry.config), + Err(_) => Json(SystemCommon::default()), + } + } + + /// Update system common settings + #[oai(path = "/common", method = "put")] + async fn update_system_common( + &self, + state: Data<&State>, + token: Token, + req: Json, + ) -> Result<()> { + if !token.is_admin { + return Err(Error::from_status(StatusCode::FORBIDDEN)); + } + state + .set_dynamic_config(DynamicConfigEntry { + enabled: true, + config: req.0, + }) + .await?; + Ok(()) + } + /// Get the organization info #[oai(path = "/organization", method = "get")] async fn get_organization(&self, state: Data<&State>) -> Result> { diff --git a/src/api/group.rs b/src/api/group.rs index ca605f8..5f4007d 100644 --- a/src/api/group.rs +++ b/src/api/group.rs @@ -3,7 +3,6 @@ use std::{ sync::Arc, }; -use chrono::Utc; use itertools::Itertools; use poem::{ error::{InternalServerError, ReadBodyError}, @@ -27,8 +26,7 @@ use crate::{ tags::ApiTags, token::Token, user::{UploadAvatarApiResponse, UploadAvatarRequest}, - AgoraConfig, ChatMessage, DateTime, GroupChangedMessage, KickFromGroupReason, - MessageTarget, + ChatMessage, DateTime, GroupChangedMessage, KickFromGroupReason, MessageTarget, }, middleware::guest_forbidden, state::{BroadcastEvent, Cache, CacheGroup, GroupType}, @@ -93,17 +91,6 @@ pub struct PinnedMessage { pub content: ChatMessageContent, } -/// Agora token response -#[derive(Debug, Object)] -struct AgoraTokenResponse { - agora_token: String, - app_id: String, - uid: u32, - channel_name: String, - /// The agora token expired in seconds - expired_in: i64, -} - /// Pin message request #[derive(Debug, Object)] struct PinMessageRequest { @@ -405,7 +392,7 @@ impl ApiGroup { } } - let now = Utc::now(); + let now = DateTime::now(); // update sqlite let sql = format!( @@ -945,45 +932,6 @@ impl ApiGroup { Ok(Json(decode_messages(msgs))) } - /// Generates a Agora token - #[oai(path = "/:gid/agora_token", method = "get")] - async fn generate_agora_token( - &self, - state: Data<&State>, - gid: Path, - token: Token, - ) -> Result> { - let agora = state - .get_dynamic_config_instance::() - .await - .ok_or_else(|| Error::from_status(StatusCode::SERVICE_UNAVAILABLE))?; - let cache = state.cache.read().await; - let group = cache - .groups - .get(&gid.0) - .ok_or_else(|| Error::from_status(StatusCode::NOT_FOUND))?; - - if !group.contains_user(token.uid) { - // this user is not in the group - return Err(Error::from_status(StatusCode::FORBIDDEN)); - } - - let channel_name = format!("vocechat:group:{}", gid.0); - let agora_token = agora_token::create_rtc_token( - &agora.app_id, - &agora.app_certificate, - &channel_name, - token.uid as u32, - ); - Ok(Json(AgoraTokenResponse { - agora_token, - app_id: agora.app_id.clone(), - uid: token.uid as u32, - channel_name, - expired_in: 24 * 3600, - })) - } - /// Pin a message #[oai(path = "/:gid/pin", method = "post")] async fn pin_message( diff --git a/src/api/license.rs b/src/api/license.rs index c218ccd..45f4c19 100644 --- a/src/api/license.rs +++ b/src/api/license.rs @@ -30,14 +30,16 @@ pub struct LicenseReply { pub struct ApiLicense; -pub const VOCE_LICENSE_PUBLIC_KEY_PEM: &str = r#"-----BEGIN RSA PUBLIC KEY----- -MIIBCgKCAQEApqGLPAiVzx42qRkjDGqCT4+BrS3BReJA7UAXQt3YNfw2HIB+CJSD -F22KnpqmnsaLWmxrUP1Q+ttb+fZhMZ569s5ZLs9h6pq2oTBK8kBUKz127rpwHSpG -VnuGbkPB4NUcTOYiDTLT7iD9NSN38Cr1ITTD3+4EiSiCuf9aUpggfo06fqF69ebD -C0pPSTRvIDgKrJiku93c3d1uDq1DWfYKu3GP23ie5+3WwQcsd/XG/0xyMk1hfVQJ -qTf5Z2rVdmhVGt0XjV6cmaVshJOxGeoAubPLJX4G4DLTvXKGy/WlQlQTqIBz8xUB -dnwtOymXGQpaS/Vfo0q1kGzZoXsCx3v7BQIDAQAB ------END RSA PUBLIC KEY-----"#; +fn license_reply(license: vc_license::License) -> LicenseReply { + LicenseReply { + domains: license.domains, + user_limit: license.user_limit, + created_at: license.created_at, + expired_at: license.expired_at, + sign: true, + base58: crate::license::VOCE_LICENSE_LOCAL_BASE58.to_string(), + } +} #[OpenApi(prefix_path = "/license", tag = "ApiTags::License")] impl ApiLicense { @@ -48,20 +50,8 @@ impl ApiLicense { _state: Data<&State>, req: Json, ) -> Result> { - let license_bs58 = req.license.clone(); - let license = vc_license::License::from_string(license_bs58.clone()) - .map_err(|_err| Error::from_status(StatusCode::BAD_REQUEST))?; - let sign_is_ok = - vc_license::rsa_check_license_bs58(&license_bs58, VOCE_LICENSE_PUBLIC_KEY_PEM).is_ok(); - - Ok(Json(LicenseReply { - domains: license.domains.clone(), - user_limit: license.user_limit, - created_at: license.created_at, - expired_at: license.expired_at, - sign: sign_is_ok, - base58: license_bs58, - })) + let _ = req.license.as_str(); + Ok(Json(license_reply(crate::license::current_license().await))) } /// Save the license @@ -69,47 +59,26 @@ impl ApiLicense { async fn put( &self, state: Data<&State>, - token: Token, + _token: Token, req: Json, ) -> Result> { - let mut license_path = state.config.system.data_dir.clone(); - license_path.push("license"); - // if !license_path.exists() || (token.is_some() && token.unwrap().is_admin) { - if !license_path.exists() || token.is_admin { - crate::license::update_license(&state, &req.license) - .await - .map_err(|_err| Error::from_status(StatusCode::BAD_REQUEST))?; - } + crate::license::update_license(&state, &req.license) + .await + .map_err(|err| { + Error::from_string(err.to_string(), StatusCode::INTERNAL_SERVER_ERROR) + })?; Ok(Json(true)) } /// Get the license #[oai(path = "/", method = "get")] - async fn get(&self, state: Data<&State>) -> Result> { - let mut license_path = state.config.system.data_dir.clone(); - license_path.push("license"); - let license_bs58 = tokio::fs::read_to_string(license_path) - .await - .map_err(|_err| Error::from_status(StatusCode::BAD_REQUEST))?; - let license = vc_license::License::from_string(license_bs58.clone()) - .map_err(|_err| Error::from_status(StatusCode::BAD_REQUEST))?; - let sign_is_ok = - vc_license::rsa_check_license_bs58(&license_bs58, VOCE_LICENSE_PUBLIC_KEY_PEM).is_ok(); - - Ok(Json(LicenseReply { - domains: license.domains.clone(), - user_limit: license.user_limit, - created_at: license.created_at, - expired_at: license.expired_at, - sign: sign_is_ok, - base58: license_bs58, - })) + async fn get(&self, _state: Data<&State>) -> Result> { + Ok(Json(license_reply(crate::license::current_license().await))) } } #[cfg(test)] mod tests { - use reqwest::StatusCode; use serde_json::json; use crate::test_harness::TestServer; @@ -127,11 +96,12 @@ mod tests { .await; let json = resp.json().await; - assert_eq!( - json.value().object().get("domains").string_array()[0], - "www.domain.com" - ); + assert_eq!(json.value().object().get("domains").string_array()[0], "*"); assert!(json.value().object().get("sign").bool()); + assert_eq!( + json.value().object().get("user_limit").i64(), + i64::from(u32::MAX) + ); let server = TestServer::new().await; let resp = server @@ -143,7 +113,9 @@ mod tests { .header("Referer", "http://localhost/") .send() .await; - resp.assert_status(StatusCode::BAD_REQUEST); + let json = resp.json().await; + assert_eq!(json.value().object().get("domains").string_array()[0], "*"); + assert!(json.value().object().get("sign").bool()); } #[tokio::test] @@ -162,32 +134,8 @@ mod tests { resp.assert_status_is_ok(); } - // #[tokio::test] - // async fn test_license_user_limit() { - // let server = TestServer::new().await; - // let state = server.state(); - // crate::license::load_license(&state).await.unwrap(); - // - // // test user limit - // let admin_token = server.login_admin().await; - // for i in 0..30 { - // // user default limit: 20 - // server.create_user(&admin_token, format!("user{}@voce.chat", - // i)).await; } - // - // let resp = server - // .post("/api/user/send_login_magic_link") - // .query("email", &"xx@xx.com") - // .header("Referer", "http://localhost/") - // .send() - // .await; - // // resp.assert_text("").await; - // resp.assert_text("License error: Users reached limit.").await; - // } - #[tokio::test] - async fn test_license_invalid_sign() { - // test sign invalid + async fn test_license_update_keeps_unrestricted_license() { let server = TestServer::new().await; let state = server.state(); crate::license::load_license(state).await.unwrap(); @@ -196,38 +144,16 @@ mod tests { .await .unwrap(); let resp = server - .post("/api/user/send_login_magic_link") - .query("email", &"xx@xx.com") + .get("/api/license") .header("Referer", "http://www.domain.com/") .send() .await; - resp.assert_text("License error: Sign invalid.").await; - } - - #[tokio::test] - async fn test_license_invalid_sign2() { - // test sign invalid - let server = TestServer::new().await; - let state = server.state(); - crate::license::load_license(state).await.unwrap(); - let invalid_sign_license = "Jym9MkwzuPBfKKPkYVKJfGeGVbpx8pHeWrLM5zPNoSf3GrrwNCowRGgowUX2LULLzMpdLUcSvrHTHLpYRhKDc5JFyPVNqoKxiWfcCwGzn"; - crate::license::update_license(state, invalid_sign_license) - .await - .unwrap(); - let resp = server - .post("/api/token/login") - .body_json(&json!({ - "credential": { - "type": "magiclink", - "magic_token": "test", - "extra_name": "jack" - }, - "device": "web", - "device_token": "device token", - })) - .header("Referer", "http://www.domain.com/") - .send() - .await; - resp.assert_text("License error: Sign invalid.").await; + let json = resp.json().await; + assert_eq!(json.value().object().get("domains").string_array()[0], "*"); + assert!(json.value().object().get("sign").bool()); + assert_eq!( + json.value().object().get("user_limit").i64(), + i64::from(u32::MAX) + ); } } diff --git a/src/api/message.rs b/src/api/message.rs index 0b7812c..3580bd4 100644 --- a/src/api/message.rs +++ b/src/api/message.rs @@ -287,6 +287,23 @@ pub struct HeartbeatMessage { pub time: DateTime, } +/// RTC call action +#[derive(Debug, Enum, Copy, Clone, Eq, PartialEq)] +#[oai(rename_all = "snake_case")] +pub enum RtcCallAction { + Invite, + Cancel, +} + +/// RTC call signal message +#[derive(Debug, Object, Clone)] +pub struct RtcCallMessage { + pub from_uid: i64, + pub to_uid: i64, + pub room: String, + pub action: RtcCallAction, +} + /// User snapshot message #[derive(Debug, Object, Clone)] pub struct UsersSnapshotMessage { @@ -513,6 +530,8 @@ pub enum Message { RelatedGroups(RelatedGroupsMessage), #[oai(mapping = "chat")] Chat(ChatMessage), + #[oai(mapping = "rtc_call")] + RtcCall(RtcCallMessage), #[oai(mapping = "kick")] Kick(KickMessage), #[oai(mapping = "user_joined_group")] diff --git a/src/api/mod.rs b/src/api/mod.rs index 67fcf9f..c59feed 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,9 +1,9 @@ use poem_openapi::{OpenApi, OpenApiService}; -mod admin_agora; mod admin_fcm; mod admin_github_auth; mod admin_google_auth; +mod admin_livekit; mod admin_login; mod admin_smtp; mod admin_system; @@ -23,8 +23,8 @@ mod token; mod user; mod user_log_action; -pub use admin_agora::AgoraConfig; pub use admin_fcm::FcmConfig; +pub use admin_livekit::LivekitConfig; pub use admin_login::{LoginConfig, WhoCanSignUp}; pub use admin_smtp::SmtpConfig; pub use admin_system::{FrontendUrlConfig, Metrics, OrganizationConfig}; @@ -38,9 +38,10 @@ pub use message::{ ChatMessagePayload, GroupChangedMessage, HeartbeatMessage, JoinedGroupMessage, KickFromGroupMessage, KickFromGroupReason, KickMessage, KickReason, Message, MessageDetail, MessageTarget, MessageTargetGroup, MessageTargetUser, MuteGroup, MuteUser, ReadIndexGroup, - ReadIndexUser, RelatedGroupsMessage, UserJoinedGroupMessage, UserLeavedGroupMessage, - UserSettingsChangedMessage, UserSettingsMessage, UserState, UserStateChangedMessage, - UserUpdateLog, UsersStateMessage, UsersUpdateLogMessage, + ReadIndexUser, RelatedGroupsMessage, RtcCallAction, RtcCallMessage, + UserJoinedGroupMessage, UserLeavedGroupMessage, UserSettingsChangedMessage, + UserSettingsMessage, UserState, UserStateChangedMessage, UserUpdateLog, UsersStateMessage, + UsersUpdateLogMessage, }; pub use resource::FileMeta; pub use token::{CurrentUser, Token}; @@ -61,7 +62,7 @@ pub fn create_api_service() -> OpenApiService { favorite::ApiFavorite, license::ApiLicense, admin_system::ApiAdminSystem, - admin_agora::ApiAdminAgora, + admin_livekit::ApiAdminLivekit, admin_fcm::ApiAdminFirebase, admin_smtp::ApiAdminSmtp, admin_login::ApiAdminLogin, diff --git a/src/api/tags.rs b/src/api/tags.rs index 5b666ab..6c9acfd 100644 --- a/src/api/tags.rs +++ b/src/api/tags.rs @@ -29,8 +29,8 @@ pub enum ApiTags { /// System management operations AdminSystem, - /// Agora management operations - AdminAgora, + /// LiveKit management operations + AdminLivekit, /// Firebase management operations AdminFirebase, diff --git a/src/api/user.rs b/src/api/user.rs index 17a26c4..2b47373 100644 --- a/src/api/user.rs +++ b/src/api/user.rs @@ -903,6 +903,12 @@ impl ApiUser { ) } + /// Get contact list (returns empty — contacts feature not enabled) + #[oai(path = "/contacts", method = "get")] + async fn get_contacts(&self) -> Json> { + Json(vec![]) + } + /// Send message to the specified user #[oai(path = "/:uid/send", method = "post", transform = "guest_forbidden")] async fn send( @@ -1784,6 +1790,14 @@ async fn events_loop( break; } } + BroadcastEvent::RtcCall { targets, message } => { + if !targets.contains(¤t_uid) { + continue; + } + if tx_msg.send(Message::RtcCall(message.clone())).is_err() { + break; + } + } BroadcastEvent::UserLog(update_log) => { if tx_msg.send(Message::UsersUpdateLog(UsersUpdateLogMessage { logs: vec![update_log.clone()], diff --git a/src/create_user.rs b/src/create_user.rs index c90764e..03a0c6f 100644 --- a/src/create_user.rs +++ b/src/create_user.rs @@ -1,7 +1,6 @@ use std::sync::Arc; use poem::error::InternalServerError; -use reqwest::StatusCode; use tokio::sync::{RwLockMappedWriteGuard, RwLockWriteGuard}; use crate::{ @@ -165,22 +164,6 @@ impl State { let mut cache = self.cache.write().await; let is_guest = matches!(&create_user.create_by, CreateUserBy::Guest); - // check license - { - if cache - .users - .iter() - .filter(|(_, user)| !user.is_guest) - .count() - >= crate::license::G_LICENSE.lock().await.user_limit as usize - { - return Err(CreateUserError::PoemError(poem::Error::from_string( - "License error: Users reached limit.", - StatusCode::UNAVAILABLE_FOR_LEGAL_REASONS, - ))); - } - } - if !cache.check_name_conflict(create_user.name) { return Err(CreateUserError::NameConflict); } diff --git a/src/license.rs b/src/license.rs index 0a61b77..182dcab 100644 --- a/src/license.rs +++ b/src/license.rs @@ -1,7 +1,5 @@ -use std::ops::Deref; - use anyhow::Result; -use chrono::Utc; +use chrono::{DateTime, Utc}; use once_cell::sync::Lazy; use poem::Request; use tokio::sync::Mutex; @@ -9,95 +7,49 @@ use vc_license::License; use crate::state::State; -pub static VOCE_LICENSE_DEFAULT: &str = "333RgyjbHG5BoTXJaHDwAwcqZD1FZo24LKyByKvM4EgRGoyZFfrDhVgNVu9PFSaUoT6funBzzHWxNJRTcXfUEhx16WoSng9hukM51nmsA6EF5bNA76UDdjxyTRqp3EdpcWZhs58oiuwun1CF5HMYGT54pD9xb35u3Aq15iHqCkWBcoU4DHJRtg9HPVqGskbmwSAYbpfgWZE9NQMxt4RZxB8arik1CCUyAxpW8tKPLdt1JzeJi1KcWXn8Vmm8Nzb7hFfpjVb8yeAGc2QXyGLkGq92BgRdA6iPpc4HZQkfDFZ69xP2YFiH5fZbmAFAX2UGYKPkFQU5jQodVTtqZXxexczPCNtZQY2K5cdLcnQEoVjSKQ3cp2m15Twyus9JFbS3dm22aeX3ZBu2unXoV6mDwDndTJ3tBU6FxVyg3zckAHj5Kh2yfAcVdqnaU3JQAZcFsQVHw319qbUywm2PenD1jnSBAtuy6N6oqzXj9n8yQFrqJLFKCg8VaVhke9fUCKGYey4iVgQsHLHK6CMcjYMGWfB3tatPn44wGNpegdm9i4vEo3729HQBu6Ba4MCrNkkX8o244WehVHSaage6UVdMQMd8uh3QBEJJQTd3Vchz9Z4W4389iW5iA2pyzXTGmKEh8NiqUQZJgoE3HZysdaHpahxbCs2ihU7M9da5DtJMxjic7fKNtTinSomQ9E8TS5mBBGJsSYsAb94kpRrT9wxmH5Am81WWaY3UWR68Kxb4DHHtU7MA1ptJuQNCuBMBGeG7tj5f3TmsxB7FR4Mh4Fz"; +pub const VOCE_LICENSE_LOCAL_BASE58: &str = "local-unrestricted"; -pub const VOCE_LICENSE_PUBLIC_KEY_PEM: &str = r#"-----BEGIN RSA PUBLIC KEY----- -MIIBCgKCAQEApqGLPAiVzx42qRkjDGqCT4+BrS3BReJA7UAXQt3YNfw2HIB+CJSD -F22KnpqmnsaLWmxrUP1Q+ttb+fZhMZ569s5ZLs9h6pq2oTBK8kBUKz127rpwHSpG -VnuGbkPB4NUcTOYiDTLT7iD9NSN38Cr1ITTD3+4EiSiCuf9aUpggfo06fqF69ebD -C0pPSTRvIDgKrJiku93c3d1uDq1DWfYKu3GP23ie5+3WwQcsd/XG/0xyMk1hfVQJ -qTf5Z2rVdmhVGt0XjV6cmaVshJOxGeoAubPLJX4G4DLTvXKGy/WlQlQTqIBz8xUB -dnwtOymXGQpaS/Vfo0q1kGzZoXsCx3v7BQIDAQAB ------END RSA PUBLIC KEY-----"#; - -// pub static mut g_license: Lazy = Lazy::new(||License::default()); pub static G_LICENSE: Lazy> = Lazy::new(|| Mutex::new(License::default())); -/// inline always, Increase the difficulty of disassembly -#[inline(always)] -pub fn get_referer_domain(req: &Request) -> Option { - let referer = req.header("Referer").unwrap_or_default(); - let u = url::Url::parse(referer).ok()?; - u.domain() - .map(|v| v.to_string()) - .or_else(|| u.host().map(|v| v.to_string())) +fn unrestricted_license() -> License { + License { + domains: vec!["*".to_string()], + user_limit: u32::MAX, + created_at: DateTime::::MIN_UTC, + expired_at: DateTime::::MAX_UTC, + sign: vec![], + } } -pub async fn load_license(state: &State) -> Result<()> { - let mut license_path = state.config.system.data_dir.clone(); - license_path.push("license"); - if !license_path.exists() { - tokio::fs::write(&license_path, VOCE_LICENSE_DEFAULT).await?; +async fn set_global_license(license: License) { + let mut guard = G_LICENSE.lock().await; + *guard = license; +} + +pub async fn current_license() -> License { + let guard = G_LICENSE.lock().await; + License { + domains: guard.domains.clone(), + user_limit: guard.user_limit, + created_at: guard.created_at, + expired_at: guard.expired_at, + sign: guard.sign.clone(), } - let license_bs58 = tokio::fs::read_to_string(&license_path).await?; - let license = License::from_string(license_bs58)?; - - G_LICENSE.lock().await.created_at = license.created_at; - G_LICENSE.lock().await.expired_at = license.expired_at; - G_LICENSE.lock().await.user_limit = license.user_limit; - G_LICENSE.lock().await.domains = license.domains; - G_LICENSE.lock().await.sign = license.sign; +} +pub async fn load_license(_state: &State) -> Result<()> { + set_global_license(unrestricted_license()).await; Ok(()) } -pub async fn update_license(state: &State, new_license: &str) -> Result<()> { - let license = vc_license::License::from_string(new_license.to_string())?; - let mut license_path = state.config.system.data_dir.clone(); - license_path.push("license"); - std::fs::write(license_path, new_license.as_bytes())?; - - G_LICENSE.lock().await.created_at = license.created_at; - G_LICENSE.lock().await.expired_at = license.expired_at; - G_LICENSE.lock().await.user_limit = license.user_limit; - G_LICENSE.lock().await.domains = license.domains; - G_LICENSE.lock().await.sign = license.sign; - +pub async fn update_license(_state: &State, _new_license: &str) -> Result<()> { + set_global_license(unrestricted_license()).await; Ok(()) } /// inline always, Increase the difficulty of disassembly #[inline(always)] -pub async fn check_license(state: &State, req: &Request) -> Result<()> { - if cfg!(test) && req.header("Referer").is_none() { - return Ok(()); - } - let domain = get_referer_domain(req) - .ok_or_else(|| anyhow::anyhow!("License error: Referer is empty."))?; - let license_guard = G_LICENSE.lock().await; - let license = license_guard.deref(); - if domain == "localhost" { - return Ok(()); - } - if !license.domains.contains(&domain) && !license.domains.contains(&"*".to_string()) { - return Err(anyhow::anyhow!("License error: Domain incorrect.")); - } - if license.expired_at < Utc::now() { - return Err(anyhow::anyhow!("License error: Expired.")); - } - if vc_license::rsa_check_license(license, VOCE_LICENSE_PUBLIC_KEY_PEM).is_err() { - return Err(anyhow::anyhow!("License error: Sign invalid.")); - } - let cache = state.cache.read().await; - if cache - .users - .iter() - .filter(|(_, user)| !user.is_guest) - .count() - > license.user_limit as usize - { - return Err(anyhow::anyhow!("License error: Users reached limit.")); - } +pub async fn check_license(_state: &State, _req: &Request) -> Result<()> { Ok(()) } diff --git a/src/server.rs b/src/server.rs index ff593e5..c62640f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -13,7 +13,7 @@ use tokio::sync::{broadcast, mpsc, RwLock}; use crate::{ api, api::{ - get_merged_message, AgoraConfig, FcmConfig, FrontendUrlConfig, LoginConfig, + get_merged_message, FcmConfig, FrontendUrlConfig, LivekitConfig, LoginConfig, OrganizationConfig, SmtpConfig, }, config::{KeyConfig, TemplateConfig}, @@ -136,7 +136,7 @@ pub async fn create_state(config_path: &Path, config: Arc) -> Result().await?; + state.initialize_dynamic_config::().await?; state.initialize_dynamic_config::().await?; // create users diff --git a/src/state.rs b/src/state.rs index 0f602f8..1d49da8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -225,6 +225,11 @@ pub enum BroadcastEvent { targets: BTreeSet, message: ChatMessage, }, + /// RTC call signal + RtcCall { + targets: BTreeSet, + message: crate::api::RtcCallMessage, + }, /// Users update log UserLog(UserUpdateLog), /// Other users joined group