781 lines
28 KiB
Rust
781 lines
28 KiB
Rust
use argon2::{
|
|
Argon2, PasswordHash, PasswordHasher, PasswordVerifier,
|
|
password_hash::{SaltString, rand_core::OsRng},
|
|
};
|
|
use axum::{
|
|
Json, Router,
|
|
extract::{
|
|
ConnectInfo, Path, Query, Request, State,
|
|
ws::{Message, WebSocket, WebSocketUpgrade},
|
|
},
|
|
http::{HeaderMap, StatusCode, header},
|
|
middleware::{self, Next},
|
|
response::{IntoResponse, Response},
|
|
routing::{delete, get, post},
|
|
};
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use hmac::{Hmac, Mac};
|
|
use serde::Deserialize;
|
|
use serde_json::{Value, json};
|
|
use sha2::{Digest, Sha256};
|
|
use sqlx::{Row, SqlitePool, sqlite::SqlitePoolOptions};
|
|
use std::{
|
|
collections::HashMap,
|
|
net::SocketAddr,
|
|
path::PathBuf,
|
|
sync::Arc,
|
|
time::{SystemTime, UNIX_EPOCH},
|
|
};
|
|
use tokio::sync::{Mutex, mpsc};
|
|
use tower_http::services::{ServeDir, ServeFile};
|
|
use uuid::Uuid;
|
|
|
|
const IDLE: i64 = 30 * 60;
|
|
const ABSOLUTE: i64 = 12 * 60 * 60;
|
|
fn now() -> i64 {
|
|
SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_secs() as i64
|
|
}
|
|
fn id() -> String {
|
|
Uuid::new_v4().to_string()
|
|
}
|
|
fn digest(s: &str) -> String {
|
|
hex::encode(Sha256::digest(s.as_bytes()))
|
|
}
|
|
type ApiResult<T> = Result<T, ApiError>;
|
|
#[derive(Debug)]
|
|
struct ApiError(StatusCode, &'static str);
|
|
impl IntoResponse for ApiError {
|
|
fn into_response(self) -> Response {
|
|
(self.0, Json(json!({"error": self.1}))).into_response()
|
|
}
|
|
}
|
|
impl From<sqlx::Error> for ApiError {
|
|
fn from(e: sqlx::Error) -> Self {
|
|
eprintln!("database: {e}");
|
|
Self(StatusCode::INTERNAL_SERVER_ERROR, "server")
|
|
}
|
|
}
|
|
fn bad(code: &'static str) -> ApiError {
|
|
ApiError(StatusCode::BAD_REQUEST, code)
|
|
}
|
|
fn unauthorized() -> ApiError {
|
|
ApiError(StatusCode::UNAUTHORIZED, "unauthorized")
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct App {
|
|
db: SqlitePool,
|
|
key: Arc<Vec<u8>>,
|
|
setup: Arc<String>,
|
|
origin: Arc<String>,
|
|
allowed_origins: Arc<Vec<String>>,
|
|
secure: bool,
|
|
peers: Arc<Mutex<Peers>>,
|
|
limits: Arc<Mutex<HashMap<String, (i64, u32)>>>,
|
|
hashes: Arc<tokio::sync::Semaphore>,
|
|
}
|
|
#[derive(Default)]
|
|
struct Peers {
|
|
clients: HashMap<String, Client>,
|
|
host: Option<String>,
|
|
since: Option<i64>,
|
|
}
|
|
struct Client {
|
|
session: String,
|
|
user: String,
|
|
name: String,
|
|
mode: String,
|
|
tx: mpsc::Sender<Message>,
|
|
shutdown: tokio::sync::watch::Sender<bool>,
|
|
}
|
|
#[derive(Clone)]
|
|
struct Auth {
|
|
session: String,
|
|
user: String,
|
|
name: String,
|
|
role: String,
|
|
}
|
|
impl Auth {
|
|
fn admin(&self) -> ApiResult<()> {
|
|
if self.role == "admin" {
|
|
Ok(())
|
|
} else {
|
|
Err(ApiError(StatusCode::FORBIDDEN, "forbidden"))
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn database(url: &str) -> Result<SqlitePool, sqlx::Error> {
|
|
let db = SqlitePoolOptions::new()
|
|
.max_connections(5)
|
|
.connect_with(
|
|
url.parse::<sqlx::sqlite::SqliteConnectOptions>()?
|
|
.create_if_missing(true)
|
|
.foreign_keys(true)
|
|
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
|
|
.busy_timeout(std::time::Duration::from_secs(5)),
|
|
)
|
|
.await?;
|
|
sqlx::raw_sql("CREATE TABLE IF NOT EXISTS accounts(id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE COLLATE NOCASE, password TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN ('admin','viewer')), last_login INTEGER);
|
|
CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY, token_hash TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL REFERENCES accounts(id) ON DELETE CASCADE, created INTEGER NOT NULL, last_seen INTEGER NOT NULL);
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
|
|
CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
PRAGMA optimize;").execute(&db).await?;
|
|
Ok(db)
|
|
}
|
|
fn password_valid(p: &str) -> bool {
|
|
p.chars().count() >= 12
|
|
&& p.len() <= 256
|
|
&& p.chars().any(char::is_alphabetic)
|
|
&& p.chars().any(|c| c.is_ascii_digit())
|
|
}
|
|
fn credentials(c: &Credentials) -> ApiResult<()> {
|
|
if c.username.len() < 3
|
|
|| c.username.len() > 40
|
|
|| !c
|
|
.username
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || "_.-".contains(c))
|
|
{
|
|
return Err(bad("username"));
|
|
}
|
|
if !password_valid(&c.password) {
|
|
return Err(bad("password"));
|
|
}
|
|
Ok(())
|
|
}
|
|
async fn hash_password(app: &App, password: String) -> ApiResult<String> {
|
|
let permit = app
|
|
.hashes
|
|
.clone()
|
|
.acquire_owned()
|
|
.await
|
|
.map_err(|_| bad("server"))?;
|
|
tokio::task::spawn_blocking(move || {
|
|
let _permit = permit;
|
|
Argon2::default()
|
|
.hash_password(password.as_bytes(), &SaltString::generate(&mut OsRng))
|
|
.map(|h| h.to_string())
|
|
.map_err(|_| bad("server"))
|
|
})
|
|
.await
|
|
.map_err(|_| bad("server"))?
|
|
}
|
|
async fn verify_password(app: &App, password: String, hash: String) -> bool {
|
|
let Ok(permit) = app.hashes.clone().acquire_owned().await else {
|
|
return false;
|
|
};
|
|
tokio::task::spawn_blocking(move || {
|
|
let _permit = permit;
|
|
PasswordHash::new(&hash)
|
|
.map(|h| {
|
|
Argon2::default()
|
|
.verify_password(password.as_bytes(), &h)
|
|
.is_ok()
|
|
})
|
|
.unwrap_or(false)
|
|
})
|
|
.await
|
|
.unwrap_or(false)
|
|
}
|
|
fn sign(key: &[u8], token: &str) -> String {
|
|
let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
|
|
mac.update(token.as_bytes());
|
|
format!("{token}.{}", hex::encode(mac.finalize().into_bytes()))
|
|
}
|
|
fn cookie_token(headers: &HeaderMap, key: &[u8]) -> ApiResult<String> {
|
|
let raw = headers
|
|
.get(header::COOKIE)
|
|
.and_then(|h| h.to_str().ok())
|
|
.and_then(|s| {
|
|
s.split(';')
|
|
.find_map(|c| c.trim().strip_prefix("webka_session="))
|
|
})
|
|
.ok_or_else(unauthorized)?;
|
|
let (token, sig) = raw.split_once('.').ok_or_else(unauthorized)?;
|
|
let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
|
|
mac.update(token.as_bytes());
|
|
mac.verify_slice(&hex::decode(sig).map_err(|_| unauthorized())?)
|
|
.map_err(|_| unauthorized())?;
|
|
Ok(token.to_owned())
|
|
}
|
|
async fn session_by_id(app: &App, session: &str) -> ApiResult<Auth> {
|
|
let row = sqlx::query("SELECT s.id, a.id AS user_id, a.username, a.role FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.id=? AND s.last_seen>? AND s.created>?").bind(session).bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_optional(&app.db).await?.ok_or_else(unauthorized)?;
|
|
Ok(Auth {
|
|
session: row.get("id"),
|
|
user: row.get("user_id"),
|
|
name: row.get("username"),
|
|
role: row.get("role"),
|
|
})
|
|
}
|
|
async fn auth(app: &App, headers: &HeaderMap, touch: bool) -> ApiResult<Auth> {
|
|
let token = cookie_token(headers, &app.key)?;
|
|
let session: Option<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE token_hash=?")
|
|
.bind(digest(&token))
|
|
.fetch_optional(&app.db)
|
|
.await?;
|
|
let a = session_by_id(app, &session.ok_or_else(unauthorized)?).await?;
|
|
if touch {
|
|
sqlx::query("UPDATE sessions SET last_seen=? WHERE id=?")
|
|
.bind(now())
|
|
.bind(&a.session)
|
|
.execute(&app.db)
|
|
.await?;
|
|
}
|
|
Ok(a)
|
|
}
|
|
async fn guard(State(app): State<App>, req: Request, next: Next) -> Response {
|
|
let mutation = !matches!(
|
|
*req.method(),
|
|
axum::http::Method::GET | axum::http::Method::HEAD | axum::http::Method::OPTIONS
|
|
);
|
|
if (mutation || req.uri().path() == "/api/signal")
|
|
&& !req
|
|
.headers()
|
|
.get(header::ORIGIN)
|
|
.and_then(|v| v.to_str().ok())
|
|
.is_some_and(|origin| {
|
|
origin == app.origin.as_str()
|
|
|| app.allowed_origins.iter().any(|allowed| allowed == origin)
|
|
})
|
|
{
|
|
return ApiError(StatusCode::FORBIDDEN, "origin").into_response();
|
|
}
|
|
let mut res = next.run(req).await;
|
|
res.headers_mut()
|
|
.insert(header::CACHE_CONTROL, "no-store".parse().unwrap());
|
|
res.headers_mut()
|
|
.insert(header::X_CONTENT_TYPE_OPTIONS, "nosniff".parse().unwrap());
|
|
res.headers_mut()
|
|
.insert(header::REFERRER_POLICY, "no-referrer".parse().unwrap());
|
|
res.headers_mut().insert(header::CONTENT_SECURITY_POLICY, "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; media-src 'self' blob:; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self'".parse().unwrap());
|
|
res
|
|
}
|
|
async fn throttle(app: &App, key: String) -> ApiResult<()> {
|
|
let mut limits = app.limits.lock().await;
|
|
limits.retain(|_, (start, _)| now() - *start < 60);
|
|
if limits.len() >= 4096 && !limits.contains_key(&key) {
|
|
return Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"));
|
|
}
|
|
let entry = limits.entry(key).or_insert((now(), 0));
|
|
entry.1 += 1;
|
|
if entry.1 > 10 {
|
|
Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
#[derive(Deserialize)]
|
|
struct Credentials {
|
|
username: String,
|
|
password: String,
|
|
setup_token: Option<String>,
|
|
}
|
|
async fn bootstrap_status(State(app): State<App>) -> ApiResult<Json<Value>> {
|
|
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM accounts")
|
|
.fetch_one(&app.db)
|
|
.await?;
|
|
Ok(Json(json!({"needs_setup": count == 0})))
|
|
}
|
|
async fn bootstrap(
|
|
State(app): State<App>,
|
|
ConnectInfo(addr): ConnectInfo<SocketAddr>,
|
|
Json(c): Json<Credentials>,
|
|
) -> ApiResult<Response> {
|
|
throttle(&app, addr.ip().to_string()).await?;
|
|
if c.setup_token.as_deref().map(digest) != Some(digest(&app.setup)) {
|
|
return Err(ApiError(StatusCode::FORBIDDEN, "setupToken"));
|
|
}
|
|
credentials(&c)?;
|
|
let hash = hash_password(&app, c.password).await?;
|
|
let result = sqlx::query("INSERT INTO accounts(id,username,password,role) SELECT ?,?,?,'admin' WHERE NOT EXISTS (SELECT 1 FROM accounts)").bind(id()).bind(&c.username).bind(hash).execute(&app.db).await?;
|
|
if result.rows_affected() == 0 {
|
|
return Err(ApiError(StatusCode::CONFLICT, "alreadySetup"));
|
|
}
|
|
Ok(StatusCode::CREATED.into_response())
|
|
}
|
|
async fn login(
|
|
State(app): State<App>,
|
|
ConnectInfo(addr): ConnectInfo<SocketAddr>,
|
|
Json(c): Json<Credentials>,
|
|
) -> ApiResult<Response> {
|
|
throttle(&app, addr.ip().to_string()).await?;
|
|
if c.password.len() > 256 || c.username.len() > 40 {
|
|
return Err(unauthorized());
|
|
}
|
|
let row = sqlx::query("SELECT id,password FROM accounts WHERE username=? COLLATE NOCASE")
|
|
.bind(&c.username)
|
|
.fetch_optional(&app.db)
|
|
.await?;
|
|
// Unknown users perform the same expensive Argon2 operation as existing users.
|
|
let dummy: String = sqlx::query_scalar("SELECT value FROM settings WHERE key='dummy_hash'")
|
|
.fetch_one(&app.db)
|
|
.await?;
|
|
let hash = row
|
|
.as_ref()
|
|
.map(|r| r.get::<String, _>("password"))
|
|
.unwrap_or(dummy);
|
|
if !verify_password(&app, c.password, hash.clone()).await || row.is_none() {
|
|
return Err(unauthorized());
|
|
}
|
|
let row = row.unwrap();
|
|
let uid: String = row.get("id");
|
|
let token = format!("{}{}", id(), id());
|
|
sqlx::query("DELETE FROM sessions WHERE last_seen<=? OR created<=?")
|
|
.bind(now() - IDLE)
|
|
.bind(now() - ABSOLUTE)
|
|
.execute(&app.db)
|
|
.await?;
|
|
// A reset/delete that won the race with password verification must invalidate this login.
|
|
let mut transaction = app.db.begin().await?;
|
|
let inserted = sqlx::query("INSERT INTO sessions(id,token_hash,user_id,created,last_seen) SELECT ?,?,id,?,? FROM accounts WHERE id=? AND password=?")
|
|
.bind(id()).bind(digest(&token)).bind(now()).bind(now()).bind(&uid).bind(&hash)
|
|
.execute(&mut *transaction).await?;
|
|
if inserted.rows_affected() == 0 {
|
|
return Err(unauthorized());
|
|
}
|
|
sqlx::query("UPDATE accounts SET last_login=? WHERE id=?")
|
|
.bind(now())
|
|
.bind(uid)
|
|
.execute(&mut *transaction)
|
|
.await?;
|
|
transaction.commit().await?;
|
|
let cookie = format!(
|
|
"webka_session={}; Path=/; HttpOnly; SameSite=Strict; Max-Age={ABSOLUTE}{}",
|
|
sign(&app.key, &token),
|
|
if app.secure { "; Secure" } else { "" }
|
|
);
|
|
Ok(([(header::SET_COOKIE, cookie)], Json(json!({"ok":true}))).into_response())
|
|
}
|
|
async fn me(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
|
|
let a = auth(&app, &headers, true).await?;
|
|
Ok(Json(
|
|
json!({"id":a.user,"username":a.name,"role":a.role,"session_id":a.session}),
|
|
))
|
|
}
|
|
async fn logout(State(app): State<App>, headers: HeaderMap) -> ApiResult<Response> {
|
|
let a = auth(&app, &headers, false).await?;
|
|
revoke(&app, &a.session).await?;
|
|
Ok((
|
|
[(
|
|
header::SET_COOKIE,
|
|
"webka_session=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0",
|
|
)],
|
|
StatusCode::NO_CONTENT,
|
|
)
|
|
.into_response())
|
|
}
|
|
async fn accounts(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
let rows = sqlx::query("SELECT a.id,a.username,a.role,a.last_login,(SELECT COUNT(*) FROM sessions s WHERE s.user_id=a.id AND s.last_seen>? AND s.created>?) AS active_sessions FROM accounts a ORDER BY a.role,a.username").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
|
|
Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"role":r.get::<String,_>("role"),"last_login":r.get::<Option<i64>,_>("last_login"),"active_sessions":r.get::<i64,_>("active_sessions")})).collect::<Vec<_>>())))
|
|
}
|
|
async fn create_account(
|
|
State(app): State<App>,
|
|
headers: HeaderMap,
|
|
Json(c): Json<Credentials>,
|
|
) -> ApiResult<StatusCode> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
credentials(&c)?;
|
|
let hash = hash_password(&app, c.password).await?;
|
|
let result =
|
|
sqlx::query("INSERT INTO accounts(id,username,password,role) VALUES(?,?,?,'viewer')")
|
|
.bind(id())
|
|
.bind(c.username)
|
|
.bind(hash)
|
|
.execute(&app.db)
|
|
.await;
|
|
match result {
|
|
Ok(_) => Ok(StatusCode::CREATED),
|
|
Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
|
|
Err(ApiError(StatusCode::CONFLICT, "duplicate"))
|
|
}
|
|
Err(e) => Err(e.into()),
|
|
}
|
|
}
|
|
async fn delete_account(
|
|
State(app): State<App>,
|
|
headers: HeaderMap,
|
|
Path(uid): Path<String>,
|
|
) -> ApiResult<StatusCode> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
let mut tx = app.db.begin().await?;
|
|
let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
|
|
.bind(&uid)
|
|
.fetch_all(&mut *tx)
|
|
.await?;
|
|
let result = sqlx::query("DELETE FROM accounts WHERE id=? AND role='viewer'")
|
|
.bind(&uid)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
if result.rows_affected() == 0 {
|
|
return Err(bad("protectedAccount"));
|
|
}
|
|
tx.commit().await?;
|
|
for sid in sessions {
|
|
disconnect_session(&app, &sid).await;
|
|
}
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
#[derive(Deserialize)]
|
|
struct PasswordChange {
|
|
password: String,
|
|
}
|
|
async fn reset_password(
|
|
State(app): State<App>,
|
|
headers: HeaderMap,
|
|
Path(uid): Path<String>,
|
|
Json(c): Json<PasswordChange>,
|
|
) -> ApiResult<StatusCode> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
if !password_valid(&c.password) {
|
|
return Err(bad("password"));
|
|
}
|
|
let hash = hash_password(&app, c.password).await?;
|
|
let mut tx = app.db.begin().await?;
|
|
let result = sqlx::query("UPDATE accounts SET password=? WHERE id=?")
|
|
.bind(hash)
|
|
.bind(&uid)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
if result.rows_affected() == 0 {
|
|
return Err(bad("notFound"));
|
|
}
|
|
let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
|
|
.bind(&uid)
|
|
.fetch_all(&mut *tx)
|
|
.await?;
|
|
sqlx::query("DELETE FROM sessions WHERE user_id=?")
|
|
.bind(&uid)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
tx.commit().await?;
|
|
for sid in sessions {
|
|
disconnect_session(&app, &sid).await;
|
|
}
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
async fn sessions(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
let rows = sqlx::query("SELECT s.id,s.created,s.last_seen,a.username FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.last_seen>? AND s.created>? ORDER BY s.last_seen DESC").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
|
|
Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"created":r.get::<i64,_>("created"),"last_seen":r.get::<i64,_>("last_seen")})).collect::<Vec<_>>())))
|
|
}
|
|
async fn revoke(app: &App, sid: &str) -> ApiResult<()> {
|
|
sqlx::query("DELETE FROM sessions WHERE id=?")
|
|
.bind(sid)
|
|
.execute(&app.db)
|
|
.await?;
|
|
disconnect_session(app, sid).await;
|
|
Ok(())
|
|
}
|
|
async fn revoke_session(
|
|
State(app): State<App>,
|
|
headers: HeaderMap,
|
|
Path(sid): Path<String>,
|
|
) -> ApiResult<StatusCode> {
|
|
auth(&app, &headers, true).await?.admin()?;
|
|
revoke(&app, &sid).await?;
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
fn send(client: &Client, value: Value) {
|
|
if client
|
|
.tx
|
|
.try_send(Message::Text(value.to_string().into()))
|
|
.is_err()
|
|
{
|
|
let _ = client.shutdown.send(true);
|
|
}
|
|
}
|
|
async fn disconnect_session(app: &App, sid: &str) {
|
|
let mut peers = app.peers.lock().await;
|
|
let ids: Vec<_> = peers
|
|
.clients
|
|
.iter()
|
|
.filter(|(_, c)| c.session == sid)
|
|
.map(|(id, _)| id.clone())
|
|
.collect();
|
|
for cid in ids {
|
|
remove_peer(&mut peers, &cid);
|
|
}
|
|
}
|
|
fn remove_peer(peers: &mut Peers, cid: &str) {
|
|
let Some(client) = peers.clients.remove(cid) else {
|
|
return;
|
|
};
|
|
let _ = client.tx.try_send(Message::Close(None));
|
|
let _ = client.shutdown.send(true);
|
|
if peers.host.as_deref() == Some(cid) {
|
|
peers.host = None;
|
|
peers.since = None;
|
|
for c in peers.clients.values() {
|
|
send(c, json!({"type":"host-offline"}));
|
|
}
|
|
} else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
|
|
send(host, json!({"type":"viewer-left","peer":cid}));
|
|
}
|
|
}
|
|
async fn status(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
|
|
let a = auth(&app, &headers, true).await?;
|
|
let peers = app.peers.lock().await;
|
|
let viewers: Vec<_> = peers
|
|
.clients
|
|
.iter()
|
|
.filter(|(_, c)| c.mode == "viewer")
|
|
.map(|(id, c)| {
|
|
if a.role == "admin" {
|
|
json!({"id":id,"session_id":c.session,"username":c.name,"user_id":c.user})
|
|
} else {
|
|
json!({})
|
|
}
|
|
})
|
|
.collect();
|
|
Ok(Json(
|
|
json!({"live":peers.host.is_some(),"since":peers.since,"viewer_count":viewers.len(),"viewers":if a.role=="admin" {viewers} else {vec![]},"local_url":format!("{}/watch",app.origin)}),
|
|
))
|
|
}
|
|
#[derive(Deserialize)]
|
|
struct SignalQuery {
|
|
mode: Option<String>,
|
|
}
|
|
async fn signal(
|
|
State(app): State<App>,
|
|
headers: HeaderMap,
|
|
Query(q): Query<SignalQuery>,
|
|
ws: WebSocketUpgrade,
|
|
) -> ApiResult<Response> {
|
|
let a = auth(&app, &headers, true).await?;
|
|
let mode = q.mode.unwrap_or("viewer".into());
|
|
if mode != "host" && mode != "viewer" {
|
|
return Err(bad("mode"));
|
|
}
|
|
if mode == "host" {
|
|
a.admin()?;
|
|
}
|
|
Ok(ws
|
|
.max_message_size(64 * 1024)
|
|
.max_frame_size(64 * 1024)
|
|
.on_upgrade(move |socket| connection(app, a, mode, socket)))
|
|
}
|
|
async fn connection(app: App, a: Auth, mode: String, socket: WebSocket) {
|
|
let cid = id();
|
|
let (tx, mut rx) = mpsc::channel(64);
|
|
let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
|
|
let (mut sink, mut source) = socket.split();
|
|
{
|
|
let mut peers = app.peers.lock().await;
|
|
if peers.clients.len() >= 32
|
|
|| (mode == "host" && peers.host.is_some())
|
|
|| peers.clients.values().any(|c| c.session == a.session)
|
|
{
|
|
drop(peers);
|
|
let _ = tokio::time::timeout(
|
|
std::time::Duration::from_secs(2),
|
|
sink.send(Message::Text(
|
|
json!({"type":"error","error":"hostBusy"})
|
|
.to_string()
|
|
.into(),
|
|
)),
|
|
)
|
|
.await;
|
|
return;
|
|
}
|
|
peers.clients.insert(
|
|
cid.clone(),
|
|
Client {
|
|
session: a.session.clone(),
|
|
user: a.user,
|
|
name: a.name,
|
|
mode: mode.clone(),
|
|
tx,
|
|
shutdown: shutdown_tx,
|
|
},
|
|
);
|
|
if mode == "host" {
|
|
peers.host = Some(cid.clone());
|
|
peers.since = Some(now());
|
|
for (vid, c) in &peers.clients {
|
|
if c.mode == "viewer" {
|
|
send(
|
|
peers.clients.get(&cid).unwrap(),
|
|
json!({"type":"viewer-joined","peer":vid}),
|
|
);
|
|
send(c, json!({"type":"host-online"}));
|
|
}
|
|
}
|
|
} else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
|
|
send(host, json!({"type":"viewer-joined","peer":cid}));
|
|
}
|
|
send(
|
|
peers.clients.get(&cid).unwrap(),
|
|
json!({"type":"welcome","peer":cid,"live":peers.host.is_some()}),
|
|
);
|
|
}
|
|
let mut tick = tokio::time::interval(std::time::Duration::from_secs(2));
|
|
let mut count = 0;
|
|
let mut window = now();
|
|
loop {
|
|
tokio::select! {
|
|
_ = shutdown_rx.changed() => break,
|
|
_ = tick.tick() => { if session_by_id(&app,&a.session).await.is_err() { break; } }
|
|
msg = rx.recv() => { match msg { Some(msg) => { let close = matches!(msg, Message::Close(_)); if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(msg)).await.is_ok_and(|result| result.is_ok()) || close { break; } }, None => break } }
|
|
msg = source.next() => {
|
|
match msg {
|
|
Some(Ok(Message::Text(text))) => {
|
|
if now() != window { window = now(); count = 0; } count += 1; if count > 120 { break; }
|
|
if session_by_id(&app,&a.session).await.is_err() { break; }
|
|
let Ok(v) = serde_json::from_str::<Value>(&text) else { continue; };
|
|
let Some(kind) = v["type"].as_str() else { continue; };
|
|
let peers = app.peers.lock().await;
|
|
if !peers.clients.contains_key(&cid) { break; }
|
|
let Some(target_id) = v["target"].as_str() else { continue; };
|
|
let Some(target) = peers.clients.get(target_id) else { continue; };
|
|
let allowed = if mode == "host" { peers.host.as_deref() == Some(&cid) && target.mode=="viewer" && matches!(kind,"offer"|"ice") } else { peers.host.as_deref()==Some(target_id) && matches!(kind,"answer"|"ice"|"renegotiate") };
|
|
if allowed { send(target,json!({"type":kind,"peer":cid,"data":v["data"]})); }
|
|
}
|
|
Some(Ok(Message::Ping(data))) => { if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(Message::Pong(data))).await.is_ok_and(|result| result.is_ok()) { break; } }
|
|
Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
remove_peer(&mut *app.peers.lock().await, &cid);
|
|
let _ = tokio::time::timeout(std::time::Duration::from_secs(2), sink.close()).await;
|
|
}
|
|
fn router(app: App, assets: PathBuf) -> Router {
|
|
Router::new()
|
|
.route("/api/bootstrap", get(bootstrap_status).post(bootstrap))
|
|
.route("/api/login", post(login))
|
|
.route("/api/logout", post(logout))
|
|
.route("/api/session", get(me))
|
|
.route("/api/accounts", get(accounts).post(create_account))
|
|
.route("/api/accounts/{id}", delete(delete_account))
|
|
.route("/api/accounts/{id}/password", post(reset_password))
|
|
.route("/api/sessions", get(sessions))
|
|
.route("/api/sessions/{id}", delete(revoke_session))
|
|
.route("/api/status", get(status))
|
|
.route("/api/signal", get(signal))
|
|
.fallback_service(
|
|
ServeDir::new(&assets).not_found_service(ServeFile::new(assets.join("index.html"))),
|
|
)
|
|
.layer(axum::extract::DefaultBodyLimit::max(8192))
|
|
.layer(middleware::from_fn_with_state(app.clone(), guard))
|
|
.with_state(app)
|
|
}
|
|
fn secret_file(path: &std::path::Path) -> std::io::Result<String> {
|
|
use std::io::Write;
|
|
if path.exists() {
|
|
return std::fs::read_to_string(path).map(|s| s.trim().to_owned());
|
|
}
|
|
let mut options = std::fs::OpenOptions::new();
|
|
options.write(true).create_new(true);
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::fs::OpenOptionsExt;
|
|
options.mode(0o600);
|
|
}
|
|
let mut file = options.open(path)?;
|
|
let secret = format!("{}{}", id(), id());
|
|
file.write_all(secret.as_bytes())?;
|
|
Ok(secret)
|
|
}
|
|
#[tokio::main]
|
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
let data = PathBuf::from(std::env::var("WEBKA_DATA").unwrap_or("data".into()));
|
|
std::fs::create_dir_all(&data)?;
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::fs::PermissionsExt;
|
|
std::fs::set_permissions(&data, std::fs::Permissions::from_mode(0o700))?;
|
|
}
|
|
let key = secret_file(&data.join("session-key"))?;
|
|
let setup = secret_file(&data.join("setup-token"))?;
|
|
let origin = std::env::var("WEBKA_ORIGIN")
|
|
.unwrap_or("http://localhost:3000".into())
|
|
.trim_end_matches('/')
|
|
.to_owned();
|
|
let addr: SocketAddr = std::env::var("WEBKA_BIND")
|
|
.unwrap_or("127.0.0.1:3001".into())
|
|
.parse()?;
|
|
let app = App {
|
|
db: database(&format!("sqlite://{}", data.join("webka.db").display())).await?,
|
|
key: Arc::new(key.into_bytes()),
|
|
setup: Arc::new(setup),
|
|
secure: origin.starts_with("https://"),
|
|
allowed_origins: Arc::new(
|
|
std::env::var("WEBKA_ALLOWED_ORIGINS")
|
|
.unwrap_or_default()
|
|
.split(',')
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty())
|
|
.map(str::to_owned)
|
|
.collect(),
|
|
),
|
|
origin: Arc::new(origin),
|
|
peers: Default::default(),
|
|
limits: Default::default(),
|
|
hashes: Arc::new(tokio::sync::Semaphore::new(2)),
|
|
};
|
|
let dummy = hash_password(&app, id())
|
|
.await
|
|
.map_err(|_| "hash initialization failed")?;
|
|
sqlx::query("INSERT OR IGNORE INTO settings(key,value) VALUES('dummy_hash',?)")
|
|
.bind(dummy)
|
|
.execute(&app.db)
|
|
.await?;
|
|
let mdns = if std::env::var("WEBKA_MDNS").as_deref() == Ok("1") {
|
|
match (mdns_sd::ServiceDaemon::new(), local_ip_address::local_ip()) {
|
|
(Ok(daemon), Ok(ip)) => {
|
|
let port: u16 = std::env::var("WEBKA_PUBLIC_PORT")
|
|
.unwrap_or(if app.secure { "443" } else { "3000" }.into())
|
|
.parse()?;
|
|
let service = mdns_sd::ServiceInfo::new(
|
|
"_webka._tcp.local.",
|
|
"Webka",
|
|
"webka.local.",
|
|
ip,
|
|
port,
|
|
[
|
|
("path", "/watch"),
|
|
("scheme", if app.secure { "https" } else { "http" }),
|
|
]
|
|
.as_slice(),
|
|
)?;
|
|
daemon.register(service)?;
|
|
Some(daemon)
|
|
}
|
|
_ => {
|
|
eprintln!("mDNS unavailable; use the configured local URL.");
|
|
None
|
|
}
|
|
}
|
|
} else {
|
|
None
|
|
};
|
|
let listener = tokio::net::TcpListener::bind(addr).await?;
|
|
println!(
|
|
"Webka listening at {addr}. Open {}. First-run setup token: {}",
|
|
app.origin,
|
|
data.join("setup-token").display()
|
|
);
|
|
let assets = std::env::var("WEBKA_ASSETS").unwrap_or("frontend/.output/public".into());
|
|
axum::serve(
|
|
listener,
|
|
router(app, assets.into()).into_make_service_with_connect_info::<SocketAddr>(),
|
|
)
|
|
.with_graceful_shutdown(async {
|
|
let _ = tokio::signal::ctrl_c().await;
|
|
})
|
|
.await?;
|
|
if let Some(daemon) = mdns {
|
|
let _ = daemon.shutdown();
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests;
|