eba4fd1
use std::collections::HashSet;
use auth::share::crypto::{hash_password, validate_password};
use auth::share::validate::{validate_email, validate_username};
use export::pagination::Page;
use models::space::spaces;
use models::users::user::ProfileVisibility;
use models::users::user_sessions::SessionKind;
use models::users::user_suspensions::{self, SuspensionType};
use models::users::{
user, user_access_keys, user_deletion_requests, user_emails, user_profiles,
user_recovery_codes, user_sessions, user_ssh_keys,
};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, DbErr, EntityTrait, IntoActiveModel, PaginatorTrait,
QueryFilter, QueryOrder, QuerySelect, Select, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::AdminError;
use crate::services::service::AdminService;
use crate::share::roles::{AdminActor, AdminRole};
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListUsersQuery {
pub q: Option<String>,
pub status: Option<String>,
pub limit: Option<u64>,
pub offset: Option<u64>,
pub sort: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSummary {
pub uid: Uuid,
pub username: String,
pub display_name: String,
pub avatar_url: String,
pub is_active: bool,
pub is_restricted: bool,
pub prohibit_login: bool,
pub bot: bool,
pub system: bool,
pub suspended: bool,
pub created_at: OffsetDateTime,
pub last_login_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserEmailDto {
pub id: Uuid,
pub email: String,
pub is_primary: bool,
pub is_activated: bool,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSessionDto {
pub id: Uuid,
pub kind: String,
pub last_seen_at: OffsetDateTime,
pub created_at: OffsetDateTime,
pub expires_at: OffsetDateTime,
pub is_revoked: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSpaceDto {
pub id: Uuid,
pub name: String,
pub lower_name: String,
pub is_active: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminSuspensionDto {
pub id: Uuid,
pub suspension_type: String,
pub reason: Option<String>,
pub starts_at: OffsetDateTime,
pub ends_at: Option<OffsetDateTime>,
pub revoked_at: Option<OffsetDateTime>,
pub suspended_by: Option<Uuid>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserDetail {
pub user: AdminUserSummary,
pub emails: Vec<AdminUserEmailDto>,
pub sessions: Vec<AdminUserSessionDto>,
pub spaces: Vec<AdminUserSpaceDto>,
pub suspensions: Vec<AdminSuspensionDto>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct UpdateUserCmd {
#[serde(default)]
pub display_name: Option<String>,
#[serde(default)]
pub is_active: Option<bool>,
#[serde(default)]
pub is_restricted: Option<bool>,
#[serde(default)]
pub prohibit_login: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct SuspendUserCmd {
pub suspension_type: SuspensionType,
#[serde(default)]
pub reason: Option<String>,
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_datetime"
)]
pub ends_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ResetUserPasswordCmd {
pub new_password: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RevokeSessionsResp {
pub revoked: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct DeleteUserResp {
pub user_id: Uuid,
pub anonymized: bool,
}
fn user_summary(row: &user::Model, suspended: bool) -> AdminUserSummary {
AdminUserSummary {
uid: row.uid,
username: row.username.clone(),
display_name: row.display_name.clone(),
avatar_url: export::resolve_user_avatar_url(row.uid, &row.avatar_url),
is_active: row.is_active,
is_restricted: row.is_restricted,
prohibit_login: row.prohibit_login,
bot: row.bot,
system: row.system,
suspended,
created_at: row.created_at,
last_login_at: row.last_login_at,
}
}
impl AdminService {
pub async fn list_users(
&self,
query: ListUsersQuery,
) -> Result<Page<AdminUserSummary>, AdminError> {
let limit = query.limit.unwrap_or(50).clamp(1, 100);
let offset = query.offset.unwrap_or(0);
let select = self.users_select(&query).await?;
let total = select.clone().count(self.db().read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let ids: Vec<Uuid> = rows.iter().map(|row| row.uid).collect();
let suspended = self.active_suspension_ids(&ids).await?;
let items = rows
.iter()
.map(|row| user_summary(row, suspended.contains(&row.uid)))
.collect();
Ok(Page {
items,
total,
limit,
offset,
})
}
async fn active_suspension_ids(&self, ids: &[Uuid]) -> Result<HashSet<Uuid>, AdminError> {
if ids.is_empty() {
return Ok(HashSet::new());
}
let now = OffsetDateTime::now_utc();
let rows = user_suspensions::Entity::find()
.filter(user_suspensions::Column::UserId.is_in(ids.to_vec()))
.filter(user_suspensions::Column::RevokedAt.is_null())
.filter(user_suspensions::Column::StartsAt.lte(now))
.filter(
Condition::any()
.add(user_suspensions::Column::EndsAt.is_null())
.add(user_suspensions::Column::EndsAt.gt(now)),
)
.all(self.db().read())
.await?;
Ok(rows.into_iter().map(|row| row.user_id).collect())
}
pub async fn get_user(&self, uid: Uuid) -> Result<AdminUserDetail, AdminError> {
let row = user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let suspended = self.active_suspension_ids(&[uid]).await?;
let emails = user_emails::Entity::find()
.filter(user_emails::Column::UserId.eq(uid))
.filter(user_emails::Column::DeletedAt.is_null())
.order_by_desc(user_emails::Column::IsPrimary)
.all(self.db().read())
.await?
.into_iter()
.map(|email| AdminUserEmailDto {
id: email.id,
email: email.email,
is_primary: email.is_primary,
is_activated: email.is_activated,
created_at: email.created_at,
})
.collect();
let sessions = user_sessions::Entity::find()
.filter(user_sessions::Column::UserId.eq(uid))
.order_by_desc(user_sessions::Column::LastSeenAt)
.limit(50)
.all(self.db().read())
.await?
.into_iter()
.map(|session| AdminUserSessionDto {
id: session.id,
kind: format!("{:?}", session.kind).to_ascii_lowercase(),
last_seen_at: session.last_seen_at,
created_at: session.created_at,
expires_at: session.expires_at,
is_revoked: session.is_revoked,
})
.collect();
let spaces = spaces::Entity::find()
.filter(spaces::Column::OwnerId.eq(uid))
.order_by_desc(spaces::Column::CreatedAt)
.limit(100)
.all(self.db().read())
.await?
.into_iter()
.map(|space| AdminUserSpaceDto {
id: space.id,
name: space.name,
lower_name: space.lower_name,
is_active: space.is_active,
})
.collect();
let suspensions = user_suspensions::Entity::find()
.filter(user_suspensions::Column::UserId.eq(uid))
.order_by_desc(user_suspensions::Column::CreatedAt)
.limit(50)
.all(self.db().read())
.await?
.into_iter()
.map(|row| AdminSuspensionDto {
id: row.id,
suspension_type: match row.suspension_type {
SuspensionType::Temporary => "temporary",
SuspensionType::Permanent => "permanent",
}
.to_string(),
reason: row.reason,
starts_at: row.starts_at,
ends_at: row.ends_at,
revoked_at: row.revoked_at,
suspended_by: row.suspended_by,
})
.collect();
Ok(AdminUserDetail {
user: user_summary(&row, suspended.contains(&uid)),
emails,
sessions,
spaces,
suspensions,
})
}
pub async fn update_user(
&self,
actor: &AdminActor,
uid: Uuid,
cmd: UpdateUserCmd,
) -> Result<AdminUserSummary, AdminError> {
actor.require(AdminRole::Operator)?;
let row = user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let mut am: user::ActiveModel = row.clone().into();
let mut changed: Vec<String> = Vec::new();
if let Some(display_name) = cmd.display_name {
let display_name = display_name.trim().to_string();
if display_name.is_empty() {
return Err(AdminError::BadRequest(
"display_name must not be empty".to_string(),
));
}
am.display_name = Set(display_name);
changed.push("display_name".to_string());
}
if let Some(value) = cmd.is_active {
am.is_active = Set(value);
changed.push("is_active".to_string());
}
if let Some(value) = cmd.is_restricted {
am.is_restricted = Set(value);
changed.push("is_restricted".to_string());
}
if let Some(value) = cmd.prohibit_login {
am.prohibit_login = Set(value);
changed.push("prohibit_login".to_string());
}
if changed.is_empty() {
let suspended = self.active_suspension_ids(&[uid]).await?;
return Ok(user_summary(&am_as_model(&row), suspended.contains(&uid)));
}
am.updated_at = Set(OffsetDateTime::now_utc());
am.version = Set(row.version + 1);
let row = am.update(self.db().write()).await?;
let suspended = self.active_suspension_ids(&[uid]).await?;
Ok(user_summary(&row, suspended.contains(&uid)))
}
pub async fn suspend_user(
&self,
actor: &AdminActor,
uid: Uuid,
cmd: SuspendUserCmd,
) -> Result<AdminSuspensionDto, AdminError> {
actor.require(AdminRole::Operator)?;
let row = user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let now = OffsetDateTime::now_utc();
let ends_at = match cmd.suspension_type {
SuspensionType::Temporary => {
let ends_at = cmd.ends_at.ok_or_else(|| {
AdminError::BadRequest("temporary suspension requires ends_at".to_string())
})?;
if ends_at <= now {
return Err(AdminError::BadRequest(
"ends_at must be in the future".to_string(),
));
}
Some(ends_at)
}
SuspensionType::Permanent => None,
};
let reason = cmd
.reason
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty());
let suspension = user_suspensions::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(uid),
suspension_type: Set(cmd.suspension_type),
reason: Set(reason.clone()),
suspended_by: Set(Some(actor.id)),
starts_at: Set(now),
ends_at: Set(ends_at),
revoked_at: Set(None),
revoked_by: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?;
let mut am: user::ActiveModel = row.into();
am.is_restricted = Set(true);
am.updated_at = Set(now);
am.update(self.db().write()).await?;
self.revoke_user_sessions(uid).await?;
Ok(AdminSuspensionDto {
id: suspension.id,
suspension_type: match suspension.suspension_type {
SuspensionType::Temporary => "temporary",
SuspensionType::Permanent => "permanent",
}
.to_string(),
reason: suspension.reason,
starts_at: suspension.starts_at,
ends_at: suspension.ends_at,
revoked_at: suspension.revoked_at,
suspended_by: suspension.suspended_by,
})
}
pub async fn revoke_suspension(
&self,
actor: &AdminActor,
uid: Uuid,
suspension_id: Uuid,
) -> Result<(), AdminError> {
actor.require(AdminRole::Operator)?;
let row = user_suspensions::Entity::find_by_id(suspension_id)
.one(self.db().read())
.await?
.filter(|row| row.user_id == uid)
.ok_or(AdminError::NotFound)?;
if row.revoked_at.is_some() {
return Ok(());
}
let now = OffsetDateTime::now_utc();
let mut am: user_suspensions::ActiveModel = row.into();
am.revoked_at = Set(Some(now));
am.revoked_by = Set(Some(actor.id));
am.updated_at = Set(now);
am.update(self.db().write()).await?;
let remaining = self.active_suspension_ids(&[uid]).await?;
if !remaining.contains(&uid)
&& let Some(user_row) = user::Entity::find_by_id(uid).one(self.db().write()).await?
{
let mut um: user::ActiveModel = user_row.into();
um.is_restricted = Set(false);
um.updated_at = Set(now);
um.update(self.db().write()).await?;
}
Ok(())
}
pub async fn reset_user_password(
&self,
actor: &AdminActor,
uid: Uuid,
cmd: ResetUserPasswordCmd,
) -> Result<(), AdminError> {
actor.require(AdminRole::Operator)?;
auth::share::crypto::validate_password(
&cmd.new_password,
&self.config().auth.password_policy,
)?;
let row = user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let now = OffsetDateTime::now_utc();
let hash = auth::share::crypto::hash_password(&cmd.new_password)?;
let mut am: user::ActiveModel = row.into();
am.passwd_hash = Set(Some(hash));
am.password_last_changed_at = Set(Some(now));
am.passwd_version = Set(Some(now.unix_timestamp()));
am.updated_at = Set(now);
am.update(self.db().write()).await?;
self.revoke_user_sessions(uid).await?;
Ok(())
}
pub async fn revoke_user_sessions(&self, uid: Uuid) -> Result<u64, AdminError> {
let now = OffsetDateTime::now_utc();
let result = user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::IsRevoked, Expr::value(true))
.col_expr(user_sessions::Column::RevokedAt, Expr::value(Some(now)))
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(now))
.filter(user_sessions::Column::UserId.eq(uid))
.filter(user_sessions::Column::IsRevoked.eq(false))
.exec(self.db().write())
.await?;
Ok(result.rows_affected)
}
pub async fn admin_revoke_user_sessions(
&self,
actor: &AdminActor,
uid: Uuid,
) -> Result<RevokeSessionsResp, AdminError> {
actor.require(AdminRole::Operator)?;
user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let revoked = self.revoke_user_sessions(uid).await?;
Ok(RevokeSessionsResp { revoked })
}
pub async fn admin_delete_user(
&self,
actor: &AdminActor,
uid: Uuid,
) -> Result<DeleteUserResp, AdminError> {
actor.require(AdminRole::SuperAdmin)?;
let Some(user_row) = user::Entity::find_by_id(uid).one(self.db().write()).await? else {
return Ok(DeleteUserResp {
user_id: uid,
anonymized: false,
});
};
let now = OffsetDateTime::now_utc();
let anon_name = format!("deleted-{}", &uid.simple().to_string()[..8]);
let mut active: user::ActiveModel = user_row.into_active_model();
active.username = Set(anon_name.clone());
active.lower_name = Set(anon_name.clone());
active.display_name = Set("deleted user".to_string());
active.avatar_url = Set(String::new());
active.is_active = Set(false);
active.prohibit_login = Set(true);
active.passwd_hash = Set(None);
active.password_last_changed_at = Set(None);
active.passwd_version = Set(Some(now.unix_timestamp()));
active.totp_encrypted_secret = Set(None);
active.totp_enabled = Set(false);
active.timezone = Set(None);
active.locale = Set(None);
active.language = Set(None);
active.updated_at = Set(now);
active.update(self.db().write()).await?;
if let Some(profile) = user_profiles::Entity::find_by_id(uid)
.one(self.db().write())
.await?
{
let mut am: user_profiles::ActiveModel = profile.into();
am.real_name = Set(None);
am.bio = Set(None);
am.pronouns = Set(None);
am.website = Set(None);
am.location = Set(None);
am.company = Set(None);
am.date_of_birth = Set(None);
am.banner_url = Set(None);
am.banner_hash = Set(None);
am.banner_color = Set(None);
am.accent_color = Set(None);
am.custom_status_emoji = Set(None);
am.custom_status_text = Set(None);
am.custom_status_expires_at = Set(None);
am.updated_at = Set(now);
am.update(self.db().write()).await?;
}
user_emails::Entity::update_many()
.col_expr(
user_emails::Column::Email,
Expr::value(format!("{anon_name}@deleted.invalid")),
)
.col_expr(
user_emails::Column::LowerEmail,
Expr::value(format!("{anon_name}@deleted.invalid")),
)
.col_expr(user_emails::Column::DeletedAt, Expr::value(Some(now)))
.col_expr(user_emails::Column::UpdatedAt, Expr::value(now))
.filter(user_emails::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
user_sessions::Entity::update_many()
.col_expr(user_sessions::Column::IsRevoked, Expr::value(true))
.col_expr(user_sessions::Column::RevokedAt, Expr::value(Some(now)))
.col_expr(user_sessions::Column::UpdatedAt, Expr::value(now))
.filter(user_sessions::Column::UserId.eq(uid))
.filter(user_sessions::Column::IsRevoked.eq(false))
.exec(self.db().write())
.await?;
user_access_keys::Entity::delete_many()
.filter(user_access_keys::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
user_ssh_keys::Entity::delete_many()
.filter(user_ssh_keys::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
user_recovery_codes::Entity::delete_many()
.filter(user_recovery_codes::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
user_deletion_requests::Entity::update_many()
.col_expr(
user_deletion_requests::Column::Status,
Expr::value("completed"),
)
.col_expr(
user_deletion_requests::Column::CompletedAt,
Expr::value(Some(now)),
)
.col_expr(user_deletion_requests::Column::UpdatedAt, Expr::value(now))
.filter(user_deletion_requests::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
Ok(DeleteUserResp {
user_id: uid,
anonymized: true,
})
}
}
fn am_as_model(row: &user::Model) -> user::Model {
row.clone()
}
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListUserSessionsQuery {
pub kind: Option<String>,
pub include_revoked: Option<bool>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSessionRow {
pub id: Uuid,
pub kind: String,
pub device_id: Option<Uuid>,
pub user_agent_hash: Option<String>,
pub client_ip_hash: Option<String>,
pub app_id: Option<String>,
pub last_seen_at: OffsetDateTime,
pub created_at: OffsetDateTime,
pub expires_at: OffsetDateTime,
pub is_revoked: bool,
pub revoked_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSessionPage {
pub items: Vec<AdminUserSessionRow>,
pub total: i64,
pub limit: u64,
pub offset: u64,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateUserCmd {
pub username: String,
pub email: String,
pub password: String,
#[serde(default)]
pub display_name: Option<String>,
#[serde(default)]
pub is_active: Option<bool>,
#[serde(default)]
pub email_activated: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CreateUserResp {
pub uid: Uuid,
pub username: String,
pub email: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminUserSecurityDto {
pub totp_enabled: bool,
pub recovery_codes_total: i64,
pub recovery_codes_used: i64,
pub active_sessions: i64,
pub ssh_keys: i64,
pub access_keys: i64,
pub passwd_version: i64,
pub password_last_changed_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct BulkUsersCmd {
pub action: String,
pub ids: Vec<Uuid>,
#[serde(default)]
pub reason: Option<String>,
#[serde(default)]
pub suspension_type: Option<SuspensionType>,
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_datetime"
)]
pub ends_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct BulkUserFailure {
pub uid: Uuid,
pub error: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct BulkUsersResp {
pub action: String,
pub succeeded: Vec<Uuid>,
pub failed: Vec<BulkUserFailure>,
}
impl AdminService {
async fn users_select(
&self,
query: &ListUsersQuery,
) -> Result<Select<user::Entity>, AdminError> {
let mut select = user::Entity::find();
if let Some(raw) = query
.q
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let needle = raw.to_ascii_lowercase();
let mut cond = Condition::any()
.add(user::Column::LowerName.contains(needle.clone()))
.add(user::Column::Username.contains(needle.clone()))
.add(user::Column::DisplayName.contains(raw));
let email_uids: Vec<Uuid> = user_emails::Entity::find()
.filter(user_emails::Column::LowerEmail.contains(needle))
.limit(200)
.all(self.db().read())
.await?
.into_iter()
.map(|row| row.user_id)
.collect();
if !email_uids.is_empty() {
cond = cond.add(user::Column::Uid.is_in(email_uids));
}
select = select.filter(cond);
}
select = match query.status.as_deref() {
Some("active") => select.filter(user::Column::IsActive.eq(true)),
Some("inactive") => select.filter(user::Column::IsActive.eq(false)),
Some("restricted") => select.filter(user::Column::IsRestricted.eq(true)),
Some("prohibited") => select.filter(user::Column::ProhibitLogin.eq(true)),
_ => select,
};
select = match query.sort.as_deref() {
Some("created_at_asc") => select.order_by_asc(user::Column::CreatedAt),
Some("last_login") => select.order_by_desc(user::Column::LastLoginAt),
Some("username") => select.order_by_asc(user::Column::LowerName),
_ => select.order_by_desc(user::Column::CreatedAt),
};
Ok(select)
}
pub async fn export_users(
&self,
query: ListUsersQuery,
) -> Result<Vec<AdminUserSummary>, AdminError> {
let rows = self
.users_select(&query)
.await?
.limit(5000)
.all(self.db().read())
.await?;
let ids: Vec<Uuid> = rows.iter().map(|row| row.uid).collect();
let suspended = self.active_suspension_ids(&ids).await?;
Ok(rows
.iter()
.map(|row| user_summary(row, suspended.contains(&row.uid)))
.collect())
}
pub async fn list_user_sessions(
&self,
uid: Uuid,
query: ListUserSessionsQuery,
) -> Result<AdminUserSessionPage, AdminError> {
user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let limit = query.limit.unwrap_or(50).clamp(1, 200);
let offset = query.offset.unwrap_or(0);
let mut select =
user_sessions::Entity::find().filter(user_sessions::Column::UserId.eq(uid));
if let Some(raw) = query
.kind
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let kind = match raw {
"web_login" => SessionKind::WebLogin,
"bearer_access" => SessionKind::BearerAccess,
other => {
return Err(AdminError::BadRequest(format!(
"unknown session kind: {other}"
)));
}
};
select = select.filter(user_sessions::Column::Kind.eq(kind));
}
if query.include_revoked != Some(true) {
select = select.filter(user_sessions::Column::IsRevoked.eq(false));
}
select = select.order_by_desc(user_sessions::Column::CreatedAt);
let total = select.clone().count(self.db().read()).await? as i64;
let items = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?
.into_iter()
.map(|row| AdminUserSessionRow {
id: row.id,
kind: format!("{:?}", row.kind).to_ascii_lowercase(),
device_id: row.device_id,
user_agent_hash: row.user_agent_hash,
client_ip_hash: row.client_ip_hash,
app_id: row.app_id,
last_seen_at: row.last_seen_at,
created_at: row.created_at,
expires_at: row.expires_at,
is_revoked: row.is_revoked,
revoked_at: row.revoked_at,
})
.collect();
Ok(AdminUserSessionPage {
items,
total,
limit,
offset,
})
}
pub async fn revoke_user_session(
&self,
actor: &AdminActor,
uid: Uuid,
session_id: Uuid,
) -> Result<(), AdminError> {
actor.require(AdminRole::Operator)?;
let row = user_sessions::Entity::find_by_id(session_id)
.one(self.db().read())
.await?
.filter(|row| row.user_id == uid)
.ok_or(AdminError::NotFound)?;
let now = OffsetDateTime::now_utc();
if !row.is_revoked {
let mut am: user_sessions::ActiveModel = row.into();
am.is_revoked = Set(true);
am.revoked_at = Set(Some(now));
am.updated_at = Set(now);
am.update(self.db().write()).await?;
}
Ok(())
}
pub async fn user_security(&self, uid: Uuid) -> Result<AdminUserSecurityDto, AdminError> {
let row = user::Entity::find_by_id(uid)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let codes = user_recovery_codes::Entity::find()
.filter(user_recovery_codes::Column::UserId.eq(uid))
.all(self.db().read())
.await?;
let active_sessions = user_sessions::Entity::find()
.filter(user_sessions::Column::UserId.eq(uid))
.filter(user_sessions::Column::IsRevoked.eq(false))
.count(self.db().read())
.await? as i64;
let ssh_keys = user_ssh_keys::Entity::find()
.filter(user_ssh_keys::Column::UserId.eq(uid))
.count(self.db().read())
.await? as i64;
let access_keys = user_access_keys::Entity::find()
.filter(user_access_keys::Column::UserId.eq(uid))
.count(self.db().read())
.await? as i64;
Ok(AdminUserSecurityDto {
totp_enabled: row.totp_enabled,
recovery_codes_total: codes.len() as i64,
recovery_codes_used: codes.iter().filter(|code| code.used_at.is_some()).count() as i64,
active_sessions,
ssh_keys,
access_keys,
passwd_version: row.passwd_version.unwrap_or(1),
password_last_changed_at: row.password_last_changed_at,
})
}
pub async fn clear_user_two_factor(
&self,
actor: &AdminActor,
uid: Uuid,
) -> Result<(), AdminError> {
actor.require(AdminRole::SuperAdmin)?;
let row = user::Entity::find_by_id(uid)
.one(self.db().write())
.await?
.ok_or(AdminError::NotFound)?;
let now = OffsetDateTime::now_utc();
let mut am: user::ActiveModel = row.into();
am.totp_enabled = Set(false);
am.totp_encrypted_secret = Set(None);
am.totp_last_used_at = Set(None);
am.totp_last_counter = Set(None);
am.updated_at = Set(now);
am.update(self.db().write()).await?;
user_recovery_codes::Entity::delete_many()
.filter(user_recovery_codes::Column::UserId.eq(uid))
.exec(self.db().write())
.await?;
Ok(())
}
pub async fn create_user(
&self,
actor: &AdminActor,
cmd: CreateUserCmd,
) -> Result<CreateUserResp, AdminError> {
actor.require(AdminRole::SuperAdmin)?;
let username = validate_username(cmd.username.trim())?;
let email = validate_email(&cmd.email)?;
validate_password(&cmd.password, &self.config().auth.password_policy)?;
let lower_name = username.to_lowercase();
let lower_email = email.to_lowercase();
let display_name = cmd
.display_name
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| username.clone());
let is_active = cmd.is_active.unwrap_or(true);
let email_activated = cmd.email_activated.unwrap_or(true);
let db = self.db();
let existing = user::Entity::find()
.filter(user::Column::LowerName.eq(&lower_name))
.one(db.read())
.await?;
let email_exists = user_emails::Entity::find()
.filter(user_emails::Column::LowerEmail.eq(&lower_email))
.one(db.read())
.await?;
if existing.is_some() || email_exists.is_some() {
return Err(AdminError::Conflict(
"username or email already exists".to_string(),
));
}
let password_hash = hash_password(&cmd.password)?;
let uid = Uuid::now_v7();
let now = OffsetDateTime::now_utc();
let username_for_tx = username.clone();
let lower_name_for_tx = lower_name.clone();
let display_name_for_tx = display_name.clone();
let email_for_tx = email.clone();
let lower_email_for_tx = lower_email.clone();
TransactionTrait::transaction(db.primary(), |txn| {
Box::pin(async move {
user::ActiveModel {
uid: Set(uid),
username: Set(username_for_tx),
lower_name: Set(lower_name_for_tx),
display_name: Set(display_name_for_tx),
avatar_url: Set(String::new()),
is_active: Set(is_active),
is_restricted: Set(false),
prohibit_login: Set(false),
visibility: Set(ProfileVisibility::Public),
bot: Set(false),
system: Set(false),
last_login_at: Set(None),
last_login_ip_hash: Set(None),
version: Set(1),
passwd_hash: Set(Some(password_hash)),
password_last_changed_at: Set(Some(now)),
passwd_version: Set(Some(1)),
totp_encrypted_secret: Set(None),
totp_enabled: Set(false),
totp_last_used_at: Set(None),
totp_last_counter: Set(None),
timezone: Set(None),
locale: Set(None),
language: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
user_profiles::ActiveModel {
user_id: Set(uid),
real_name: Set(None),
bio: Set(None),
pronouns: Set(None),
website: Set(None),
location: Set(None),
company: Set(None),
pronouns_privacy: Set(None),
date_of_birth: Set(None),
banner_url: Set(None),
banner_hash: Set(None),
banner_color: Set(None),
accent_color: Set(None),
timezone_privacy_flags: Set(None),
custom_status_emoji: Set(None),
custom_status_text: Set(None),
custom_status_expires_at: Set(None),
show_email: Set(false),
show_location: Set(false),
show_real_name: Set(false),
show_activity: Set(false),
privacy_agreed_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
user_emails::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(uid),
email: Set(email_for_tx),
lower_email: Set(lower_email_for_tx),
is_activated: Set(email_activated),
is_primary: Set(true),
created_at: Set(now),
updated_at: Set(now),
activated_at: Set(if email_activated { Some(now) } else { None }),
deleted_at: Set(None),
}
.insert(txn)
.await?;
Ok::<(), DbErr>(())
})
})
.await?;
Ok(CreateUserResp {
uid,
username,
email,
})
}
pub async fn bulk_users(
&self,
actor: &AdminActor,
cmd: BulkUsersCmd,
) -> Result<BulkUsersResp, AdminError> {
actor.require(AdminRole::Operator)?;
let action = cmd.action.trim().to_ascii_lowercase();
match action.as_str() {
"activate" | "deactivate" | "restrict" | "unrestrict" | "revoke_sessions"
| "suspend" | "delete" => {}
other => {
return Err(AdminError::BadRequest(format!(
"unsupported bulk action: {other}"
)));
}
}
if action == "delete" {
actor.require(AdminRole::SuperAdmin)?;
}
let mut seen: HashSet<Uuid> = HashSet::new();
let ids: Vec<Uuid> = cmd
.ids
.into_iter()
.filter(|uid| seen.insert(*uid))
.collect();
if ids.is_empty() {
return Err(AdminError::BadRequest("ids must not be empty".to_string()));
}
if ids.len() > 100 {
return Err(AdminError::BadRequest(
"at most 100 ids per request".to_string(),
));
}
let mut succeeded: Vec<Uuid> = Vec::new();
let mut failed: Vec<BulkUserFailure> = Vec::new();
for uid in ids {
let outcome = match action.as_str() {
"activate" => self
.update_user(
actor,
uid,
UpdateUserCmd {
display_name: None,
is_active: Some(true),
is_restricted: None,
prohibit_login: None,
},
)
.await
.map(|_| ()),
"deactivate" => self
.update_user(
actor,
uid,
UpdateUserCmd {
display_name: None,
is_active: Some(false),
is_restricted: None,
prohibit_login: None,
},
)
.await
.map(|_| ()),
"restrict" => self
.update_user(
actor,
uid,
UpdateUserCmd {
display_name: None,
is_active: None,
is_restricted: Some(true),
prohibit_login: None,
},
)
.await
.map(|_| ()),
"unrestrict" => self
.update_user(
actor,
uid,
UpdateUserCmd {
display_name: None,
is_active: None,
is_restricted: Some(false),
prohibit_login: None,
},
)
.await
.map(|_| ()),
"revoke_sessions" => self
.admin_revoke_user_sessions(actor, uid)
.await
.map(|_| ()),
"suspend" => {
let suspension_type = cmd.suspension_type.unwrap_or(SuspensionType::Permanent);
self.suspend_user(
actor,
uid,
SuspendUserCmd {
suspension_type,
reason: cmd.reason.clone(),
ends_at: cmd.ends_at,
},
)
.await
.map(|_| ())
}
_ => self.admin_delete_user(actor, uid).await.map(|_| ()),
};
match outcome {
Ok(()) => succeeded.push(uid),
Err(AdminError::NotFound) => failed.push(BulkUserFailure {
uid,
error: "not found".to_string(),
}),
Err(error) => failed.push(BulkUserFailure {
uid,
error: error.to_string(),
}),
}
}
Ok(BulkUsersResp {
action,
succeeded,
failed,
})
}
}