use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use anyhow::{bail, Context, Result};
use chacha20poly1305::aead::{Aead, KeyInit, Payload};
use chacha20poly1305::{XChaCha20Poly1305, XNonce};
use chrono::{Duration, Utc};
use lancedb::arrow::arrow_array::{RecordBatch, RecordBatchIterator, StringArray};
use lancedb::arrow::arrow_schema::{DataType, Field, Schema};
use serde::{Deserialize, Serialize};
pub fn env_secret(name: &str) -> Option<String> {
static READ: std::sync::OnceLock<Mutex<std::collections::HashMap<String, Option<String>>>> =
std::sync::OnceLock::new();
let cache = READ.get_or_init(Default::default);
if let Some(v) = cache.lock().ok().and_then(|c| c.get(name).cloned()) {
return v;
}
let value = read_secret(name);
if let Ok(mut c) = cache.lock() {
c.insert(name.to_string(), value.clone());
}
value
}
fn read_secret(name: &str) -> Option<String> {
if let Some(v) = std::env::var(name).ok().filter(|v| !v.trim().is_empty()) {
return Some(v.trim().to_string());
}
let path = std::env::var(format!("{name}_FILE")).ok()?;
match std::fs::read_to_string(&path) {
Ok(v) => Some(v.trim().to_string()).filter(|v| !v.is_empty()),
Err(e) => {
eprintln!("comindly: {name}_FILE ({path}) cannot be read: {e}");
None
}
}
}
pub fn public_url() -> Option<String> {
env_secret("COMINDLY_PUBLIC_URL").map(|u| u.trim_end_matches('/').to_string())
}
pub fn enabled() -> bool {
public_url().is_some()
}
pub fn admin_emails() -> Vec<String> {
env_secret("COMINDLY_ADMIN_EMAILS")
.unwrap_or_default()
.split([',', ' ', '\n'])
.map(normalize)
.filter(|e| e.contains('@'))
.collect()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum SignIn {
Email,
Code,
}
pub fn sign_in() -> Result<SignIn> {
match env_secret("COMINDLY_SIGN_IN").as_deref() {
Some("email") => {
anyhow::ensure!(
mail_configured(),
"COMINDLY_SIGN_IN=email needs COMINDLY_SMTP_URL"
);
Ok(SignIn::Email)
}
Some("code") => Ok(SignIn::Code),
Some(other) => bail!("COMINDLY_SIGN_IN is `email` or `code`, not `{other}`"),
None if mail_configured() => Ok(SignIn::Email),
None => Ok(SignIn::Code),
}
}
pub fn mode() -> SignIn {
sign_in().unwrap_or(SignIn::Code)
}
pub fn normalize(email: &str) -> String {
email.trim().to_lowercase()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Role {
Admin,
Member,
}
impl Role {
fn as_str(self) -> &'static str {
match self {
Role::Admin => "admin",
Role::Member => "member",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Via {
Session,
AdminToken,
Token,
}
#[derive(Debug, Clone, Serialize)]
pub struct Who {
pub email: String,
pub role: Role,
}
#[derive(Debug, Serialize)]
pub struct Member {
pub email: String,
pub role: Role,
pub fixed: bool,
pub created: String,
pub last_seen: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct Invite {
pub email: String,
pub role: Role,
pub created_by: String,
pub expires: String,
}
#[derive(Debug, Serialize)]
pub struct AccessToken {
pub id: String,
pub name: String,
pub created: String,
pub last_used: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct SecretInfo {
pub name: String,
pub source: &'static str,
pub hint: Option<String>,
pub updated: Option<String>,
pub updated_by: Option<String>,
}
const SESSION_DAYS: i64 = 30;
const LINK_MINUTES: i64 = 15;
const INVITE_DAYS: i64 = 7;
const MAX_FAILED: i64 = 5;
const MAX_REDEEM_FAILED: i64 = 20;
const LOCK_MINUTES: i64 = 15;
const MAX_LINKS_PER_HOUR: i64 = 5;
const CODE_ALPHABET: &[u8] = b"ABCDEFGHJKLMNPQRSTUVWXYZ23456789";
fn now() -> String {
Utc::now().to_rfc3339()
}
pub fn new_token(prefix: &str) -> String {
let mut b = [0u8; 32];
getrandom::fill(&mut b).expect("system randomness");
format!("{prefix}{}", blake3::Hash::from(b).to_hex())
}
pub fn new_code() -> String {
let mut b = [0u8; 20];
getrandom::fill(&mut b).expect("system randomness");
let chars: Vec<char> = b
.iter()
.map(|x| CODE_ALPHABET[(*x as usize) % CODE_ALPHABET.len()] as char)
.collect();
chars
.chunks(4)
.map(|c| c.iter().collect::<String>())
.collect::<Vec<_>>()
.join("-")
}
fn code_hash(code: &str) -> String {
let plain: String = code
.chars()
.filter(char::is_ascii_alphanumeric)
.map(|c| c.to_ascii_uppercase())
.collect();
hash(&format!("code:{plain}"))
}
fn hash(token: &str) -> String {
blake3::hash(token.as_bytes()).to_hex().to_string()
}
fn cipher(key: &str) -> XChaCha20Poly1305 {
let k = blake3::derive_key("comindly secrets v1", key.as_bytes());
XChaCha20Poly1305::new((&k).into())
}
fn seal(key: &str, name: &str, value: &str) -> Result<(Vec<u8>, Vec<u8>)> {
let mut nonce = [0u8; 24];
getrandom::fill(&mut nonce).context("randomness")?;
let payload = Payload {
msg: value.as_bytes(),
aad: name.as_bytes(),
};
let sealed = cipher(key)
.encrypt(XNonce::from_slice(&nonce), payload)
.map_err(|_| anyhow::anyhow!("encryption failed"))?;
Ok((sealed, nonce.to_vec()))
}
fn open(key: &str, name: &str, sealed: &[u8], nonce: &[u8]) -> Option<String> {
let payload = Payload {
msg: sealed,
aad: name.as_bytes(),
};
let plain = cipher(key)
.decrypt(XNonce::from_slice(nonce), payload)
.ok()?;
String::from_utf8(plain).ok()
}
fn hint(value: &str) -> String {
let tail: String = value
.chars()
.rev()
.take(4)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
format!("…{tail}")
}
#[derive(Clone, Serialize, Deserialize)]
struct User {
role: Role,
created: String,
last_seen: Option<String>,
}
#[derive(Clone, Serialize, Deserialize)]
struct Invitation {
email: String,
role: Role,
created_by: String,
created: String,
expires: String,
}
#[derive(Clone, Serialize, Deserialize)]
struct Link {
email: String,
created: String,
expires: String,
}
#[derive(Clone, Serialize, Deserialize)]
struct Session {
email: String,
created: String,
expires: String,
}
#[derive(Clone, Serialize, Deserialize)]
struct Token {
hash: String,
email: String,
name: String,
created: String,
last_used: Option<String>,
}
#[derive(Clone, Serialize, Deserialize)]
struct Secret {
sealed: String,
nonce: String,
hint: String,
updated: String,
updated_by: String,
}
#[derive(Clone, Serialize, Deserialize)]
struct Code {
hash: String,
created: String,
}
#[derive(Default)]
struct State {
users: BTreeMap<String, User>,
invites: HashMap<String, Invitation>,
links: HashMap<String, Link>,
sessions: HashMap<String, Session>,
tokens: HashMap<String, Token>,
secrets: BTreeMap<String, Secret>,
codes: HashMap<String, Code>,
seen: HashMap<String, String>,
config: HashMap<String, String>,
failures: Vec<(String, String)>,
}
type Change = (&'static str, String, Option<String>);
fn put(kind: &'static str, key: &str, value: &impl Serialize) -> Change {
(kind, key.to_string(), serde_json::to_string(value).ok())
}
fn del(kind: &'static str, key: &str) -> Change {
(kind, key.to_string(), None)
}
impl State {
fn load(&mut self, kind: &str, key: String, json: &str) {
fn add<T: serde::de::DeserializeOwned>(
m: &mut impl Extend<(String, T)>,
key: String,
json: &str,
) {
if let Ok(v) = serde_json::from_str(json) {
m.extend([(key, v)]);
}
}
match kind {
"user" => add(&mut self.users, key, json),
"invite" => add(&mut self.invites, key, json),
"link" => add(&mut self.links, key, json),
"session" => add(&mut self.sessions, key, json),
"token" => add(&mut self.tokens, key, json),
"secret" => add(&mut self.secrets, key, json),
"code" => add(&mut self.codes, key, json),
"seen" => add(&mut self.seen, key, json),
"config" => add(&mut self.config, key, json),
_ => {}
}
}
fn apply(&mut self, changes: &[Change]) {
for (kind, key, json) in changes {
match json {
Some(j) => self.load(kind, key.clone(), j),
None => match *kind {
"user" => drop(self.users.remove(key)),
"invite" => drop(self.invites.remove(key)),
"link" => drop(self.links.remove(key)),
"session" => drop(self.sessions.remove(key)),
"token" => drop(self.tokens.remove(key)),
"secret" => drop(self.secrets.remove(key)),
"code" => drop(self.codes.remove(key)),
"seen" => drop(self.seen.remove(key)),
"config" => drop(self.config.remove(key)),
_ => {}
},
}
}
}
}
const TABLE: &str = "server";
const CONFIG: &str = "config.toml";
const KINDS: [&str; 9] = [
"user", "invite", "link", "session", "token", "secret", "code", "seen", "config",
];
fn run<T: Send + 'static>(
f: impl std::future::Future<Output = Result<T>> + Send + 'static,
) -> Result<T> {
static RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
let rt = RT.get_or_init(|| {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.thread_name("comindly-server-store")
.enable_all()
.build()
.expect("server store runtime")
});
let (tx, rx) = std::sync::mpsc::channel();
rt.spawn(async move {
let _ = tx.send(f.await);
});
rx.recv().context("server store stopped")?
}
fn schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("kind", DataType::Utf8, false),
Field::new("key", DataType::Utf8, false),
Field::new("json", DataType::Utf8, false),
Field::new("mac", DataType::Utf8, false),
]))
}
fn mac_key(secret: &str) -> [u8; 32] {
blake3::derive_key("comindly rows v1", secret.as_bytes())
}
fn mac(key: &[u8; 32], kind: &str, k: &str, json: &str) -> String {
blake3::keyed_hash(key, format!("{kind}\0{k}\0{json}").as_bytes())
.to_hex()
.to_string()
}
fn quote(s: &str) -> String {
format!("'{}'", s.replace('\'', "''"))
}
async fn table(uri: &str) -> Result<lancedb::Table> {
let db = crate::storage::connect(uri).execute().await?;
match db.open_table(TABLE).execute().await {
Ok(t) => Ok(t),
Err(_) => match db.create_empty_table(TABLE, schema()).execute().await {
Ok(t) => Ok(t),
Err(_) => Ok(db.open_table(TABLE).execute().await?),
},
}
}
async fn read_all(uri: String) -> Result<(u64, Vec<[String; 4]>)> {
use futures::TryStreamExt;
use lancedb::query::{ExecutableQuery, QueryBase};
let tbl = table(&uri).await?;
let version = tbl.version().await?;
let mut st = tbl.query().only_if("kind != 'audit'").execute().await?;
let mut rows = Vec::new();
while let Some(b) = st.try_next().await? {
let col = |n: &str| {
b.column_by_name(n)
.and_then(|c| c.as_any().downcast_ref::<StringArray>().cloned())
};
let (Some(k), Some(key), Some(j), Some(m)) =
(col("kind"), col("key"), col("json"), col("mac"))
else {
continue;
};
for i in 0..b.num_rows() {
rows.push([k.value(i), key.value(i), j.value(i), m.value(i)].map(str::to_string));
}
}
Ok((version, rows))
}
async fn write(uri: String, key: [u8; 32], changes: Vec<Change>) -> Result<u64> {
let tbl = table(&uri).await?;
let (puts, dels): (Vec<_>, Vec<_>) = changes.into_iter().partition(|(_, _, j)| j.is_some());
if !dels.is_empty() {
let pred = dels
.iter()
.map(|(k, key, _)| format!("(kind = {} AND key = {})", quote(k), quote(key)))
.collect::<Vec<_>>()
.join(" OR ");
tbl.delete(&pred).await?;
}
if !puts.is_empty() {
let batch = RecordBatch::try_new(
schema(),
vec![
Arc::new(StringArray::from_iter_values(
puts.iter().map(|(k, _, _)| *k),
)),
Arc::new(StringArray::from_iter_values(
puts.iter().map(|(_, key, _)| key.as_str()),
)),
Arc::new(StringArray::from_iter_values(
puts.iter()
.map(|(_, _, j)| j.as_deref().unwrap_or_default()),
)),
Arc::new(StringArray::from_iter_values(puts.iter().map(
|(k, key_, j)| mac(&key, k, key_, j.as_deref().unwrap_or_default()),
))),
],
)?;
let reader = RecordBatchIterator::new(vec![Ok(batch)], schema());
let mut m = tbl.merge_insert(&["kind", "key"]);
m.when_matched_update_all(None)
.when_not_matched_insert_all();
m.execute(Box::new(reader)).await?;
}
Ok(tbl.version().await?)
}
async fn optimize(uri: String) -> Result<()> {
use lancedb::table::OptimizeAction;
let tbl = table(&uri).await?;
tbl.optimize(OptimizeAction::Compact {
options: Default::default(),
remap_options: None,
})
.await?;
tbl.optimize(OptimizeAction::Prune {
older_than: Duration::try_days(1),
delete_unverified: Some(false),
error_if_tagged_old_versions: Some(false),
})
.await?;
Ok(())
}
const REFRESH_SECONDS: i64 = 15;
pub struct Store {
uri: String,
keys: Vec<[u8; 32]>,
state: Mutex<State>,
seen: Mutex<(u64, chrono::DateTime<Utc>)>,
writing: Mutex<()>,
}
impl Store {
pub fn open(index_root: &str) -> Result<Store> {
let key =
env_secret("COMINDLY_SECRET_KEY").context("server mode needs COMINDLY_SECRET_KEY")?;
let previous = env_secret("COMINDLY_SECRET_KEY_PREVIOUS");
Store::open_keyed(index_root, &key, previous.as_deref())
}
fn open_keyed(index_root: &str, key: &str, previous: Option<&str>) -> Result<Store> {
let store = Store::attach_keyed(index_root, key, previous)?;
store.rotate_key()?;
store.prune();
let unreadable = store.unreadable_secrets();
if !unreadable.is_empty() {
eprintln!(
"comindly: {} stored secret(s) cannot be decrypted with COMINDLY_SECRET_KEY ({}); \
set COMINDLY_SECRET_KEY_PREVIOUS to the old key, or save them again",
unreadable.len(),
unreadable.join(", ")
);
}
Ok(store)
}
fn attach_keyed(index_root: &str, key: &str, previous: Option<&str>) -> Result<Store> {
let uri = format!("{}/_server", index_root.trim_end_matches('/'));
let store = Store {
uri,
keys: [Some(key), previous]
.into_iter()
.flatten()
.map(mac_key)
.collect(),
state: Mutex::new(State::default()),
seen: Mutex::new((0, Utc::now())),
writing: Mutex::new(()),
};
store.reload()?;
Ok(store)
}
fn reload(&self) -> Result<()> {
let (version, rows) = run(read_all(self.uri.clone()))?;
let mut fresh = State::default();
let (mut forged, mut resign) = (0, Vec::new());
for [kind, key, json, m] in rows {
let Some(i) = self
.keys
.iter()
.position(|k| constant_eq(&mac(k, &kind, &key, &json), &m))
else {
forged += 1;
continue;
};
if i > 0 {
if let Some(k) = KINDS.iter().find(|k| **k == kind) {
resign.push((*k, key.clone(), Some(json.clone())));
}
}
fresh.load(&kind, key, &json);
}
if forged > 0 {
eprintln!(
"comindly: ignored {forged} server row(s) not signed with COMINDLY_SECRET_KEY"
);
}
let mut st = self.lock();
fresh.failures = std::mem::take(&mut st.failures);
*st = fresh;
*self.seen.lock().unwrap_or_else(|e| e.into_inner()) = (version, Utc::now());
drop(st);
self.commit(resign)
}
fn state(&self) -> std::sync::MutexGuard<'_, State> {
let stale = {
let seen = self.seen.lock().unwrap_or_else(|e| e.into_inner());
Utc::now() - seen.1 > Duration::seconds(REFRESH_SECONDS)
};
if stale {
let uri = self.uri.clone();
let version = run(async move { Ok(table(&uri).await?.version().await?) });
let mut seen = self.seen.lock().unwrap_or_else(|e| e.into_inner());
seen.1 = Utc::now();
let changed = version.is_ok_and(|v| v != seen.0);
drop(seen);
if changed {
if let Err(e) = self.reload() {
eprintln!("comindly: re-reading the server table: {e:#}");
}
}
}
self.lock()
}
fn lock(&self) -> std::sync::MutexGuard<'_, State> {
self.state.lock().unwrap_or_else(|e| e.into_inner())
}
fn commit(&self, changes: Vec<Change>) -> Result<()> {
if changes.is_empty() {
return Ok(());
}
let version = run(write(self.uri.clone(), self.keys[0], changes.clone()))
.context("save to the server table")?;
self.lock().apply(&changes);
let mut seen = self.seen.lock().unwrap_or_else(|e| e.into_inner());
if version == seen.0 + 1 || version == seen.0 + 2 {
seen.0 = version;
}
Ok(())
}
fn prune(&self) {
let t = now();
let day = (Utc::now() - Duration::days(1)).to_rfc3339();
let mut changes = Vec::new();
{
let mut st = self.state();
changes.extend(
st.links
.iter()
.filter(|(_, l)| l.expires < t)
.map(|(k, _)| del("link", k)),
);
changes.extend(
st.invites
.iter()
.filter(|(_, i)| i.expires < t)
.map(|(k, _)| del("invite", k)),
);
changes.extend(
st.sessions
.iter()
.filter(|(_, s)| s.expires < t)
.map(|(k, _)| del("session", k)),
);
st.failures.retain(|(_, at)| *at >= day);
}
if let Err(e) = self.commit(changes) {
eprintln!("comindly: pruning the server table: {e:#}");
}
if let Err(e) = run(optimize(self.uri.clone())) {
eprintln!("comindly: optimizing the server table: {e:#}");
}
}
fn unreadable_secrets(&self) -> Vec<String> {
let Some(key) = env_secret("COMINDLY_SECRET_KEY") else {
return Vec::new();
};
self.state()
.secrets
.iter()
.filter(|(n, s)| unseal(&key, n, s).is_none())
.map(|(n, _)| n.clone())
.collect()
}
fn failures(&self, key: &str) -> i64 {
let since = (Utc::now() - Duration::minutes(LOCK_MINUTES)).to_rfc3339();
self.lock()
.failures
.iter()
.filter(|(k, at)| k == key && *at > since)
.count() as i64
}
fn fail(&self, key: &str) {
self.lock().failures.push((key.to_string(), now()));
}
fn forgive(&self, key: &str) {
self.lock().failures.retain(|(k, _)| k != key);
}
pub fn audit(&self, actor: &str, action: &str, detail: &str) {
let row =
serde_json::json!({"at": now(), "actor": actor, "action": action, "detail": detail});
let key = uuid::Uuid::new_v4().simple().to_string();
if let Err(e) = run(write(
self.uri.clone(),
self.keys[0],
vec![("audit", key, Some(row.to_string()))],
)) {
eprintln!("comindly: audit log: {e:#}");
}
}
pub fn member(&self, email: &str) -> Result<Option<Who>> {
let email = normalize(email);
if admin_emails().contains(&email) {
return Ok(Some(Who {
email,
role: Role::Admin,
}));
}
let role = self.state().users.get(&email).map(|u| u.role);
Ok(role.map(|role| Who { email, role }))
}
pub fn members(&self) -> Result<Vec<Member>> {
let admins = admin_emails();
let st = self.state();
let mut out: Vec<Member> = st
.users
.iter()
.map(|(email, u)| Member {
email: email.clone(),
role: u.role,
fixed: false,
created: u.created.clone(),
last_seen: u.last_seen.clone(),
})
.collect();
for m in &mut out {
if admins.contains(&m.email) {
m.role = Role::Admin;
m.fixed = true;
}
}
for a in admins {
if !out.iter().any(|m| m.email == a) {
out.push(Member {
last_seen: st.seen.get(&a).cloned(),
email: a,
role: Role::Admin,
fixed: true,
created: String::new(),
});
}
}
Ok(out)
}
pub fn set_role(&self, by: &str, email: &str, role: Role) -> Result<()> {
let email = normalize(email);
if admin_emails().contains(&email) {
bail!("{email} is an admin by the deployment's settings");
}
let _w = self.write_lock();
let user = self.state().users.get(&email).cloned();
let mut user = user.with_context(|| format!("{email} is not a member"))?;
user.role = role;
self.commit(vec![put("user", &email, &user)])?;
self.audit(by, "role", &format!("{email} {}", role.as_str()));
Ok(())
}
pub fn remove(&self, by: &str, email: &str) -> Result<()> {
let email = normalize(email);
if admin_emails().contains(&email) {
bail!("{email} is an admin by the deployment's settings");
}
let _w = self.write_lock();
let mut changes = vec![del("user", &email), del("code", &email)];
{
let st = self.state();
changes.extend(
st.sessions
.iter()
.filter(|(_, s)| s.email == email)
.map(|(k, _)| del("session", k)),
);
changes.extend(
st.tokens
.iter()
.filter(|(_, t)| t.email == email)
.map(|(k, _)| del("token", k)),
);
changes.extend(
st.invites
.iter()
.filter(|(_, i)| i.email == email)
.map(|(k, _)| del("invite", k)),
);
changes.extend(
st.links
.iter()
.filter(|(_, l)| l.email == email)
.map(|(k, _)| del("link", k)),
);
}
self.commit(changes)?;
let prefix = format!("code|{email}|");
self.lock()
.failures
.retain(|(k, _)| !k.starts_with(&prefix));
self.audit(by, "remove", &email);
Ok(())
}
pub fn invite(&self, by: &str, email: &str, role: Role, code: bool) -> Result<String> {
let email = normalize(email);
anyhow::ensure!(
email.contains('@') && !email.contains(char::is_whitespace),
"that is not an email address"
);
let (token, h) = if code {
let c = new_code();
let h = code_hash(&c);
(c, h)
} else {
let t = new_token("inv_");
let h = hash(&t);
(t, h)
};
let _w = self.write_lock();
let mut changes: Vec<Change> = self
.state()
.invites
.iter()
.filter(|(_, i)| i.email == email)
.map(|(k, _)| del("invite", k))
.collect();
changes.push(put(
"invite",
&h,
&Invitation {
email: email.clone(),
role,
created_by: by.to_string(),
created: now(),
expires: (Utc::now() + Duration::days(INVITE_DAYS)).to_rfc3339(),
},
));
self.commit(changes)?;
self.audit(by, "invite", &format!("{email} {}", role.as_str()));
Ok(token)
}
pub fn invites(&self) -> Result<Vec<Invite>> {
let t = now();
let mut out: Vec<(String, Invite)> = self
.state()
.invites
.values()
.filter(|i| i.expires > t)
.map(|i| {
(
i.created.clone(),
Invite {
email: i.email.clone(),
role: i.role,
created_by: i.created_by.clone(),
expires: i.expires.clone(),
},
)
})
.collect();
out.sort_by(|a, b| b.0.cmp(&a.0));
Ok(out.into_iter().map(|(_, i)| i).collect())
}
pub fn revoke_invite(&self, by: &str, email: &str) -> Result<()> {
let email = normalize(email);
let _w = self.write_lock();
let changes = self
.state()
.invites
.iter()
.filter(|(_, i)| i.email == email)
.map(|(k, _)| del("invite", k))
.collect();
self.commit(changes)?;
self.audit(by, "revoke-invite", &email);
Ok(())
}
pub fn sign_in_link(&self, email: &str) -> Result<Option<String>> {
let Some(who) = self.member(email)? else {
return Ok(None);
};
let hour = (Utc::now() - Duration::hours(1)).to_rfc3339();
let recent = self
.state()
.links
.values()
.filter(|l| l.email == who.email && l.created > hour)
.count() as i64;
if recent >= MAX_LINKS_PER_HOUR {
eprintln!("comindly: a member asked for more than {MAX_LINKS_PER_HOUR} sign-in links in an hour");
return Ok(None);
}
let token = new_token("lnk_");
let link = Link {
email: who.email,
created: now(),
expires: (Utc::now() + Duration::minutes(LINK_MINUTES)).to_rfc3339(),
};
self.commit(vec![put("link", &hash(&token), &link)])?;
Ok(Some(token))
}
pub fn redeem(&self, token: &str, client: &str) -> Result<(String, Who)> {
let token = token.trim();
let from = format!("redeem|{client}");
anyhow::ensure!(
self.failures(&from) < MAX_REDEEM_FAILED,
"Too many attempts from here; try again in {LOCK_MINUTES} minutes"
);
if let Some(admin) = env_secret("COMINDLY_ADMIN_TOKEN") {
if constant_eq(token, &admin) {
let email = admin_emails()
.into_iter()
.next()
.context("set COMINDLY_ADMIN_EMAILS to sign in with the admin token")?;
self.audit(&email, "sign-in", "admin token");
return self.session_for(&email);
}
}
let h = hash(token);
let t = now();
let _w = self.write_lock();
let link = self
.state()
.links
.get(&h)
.filter(|l| l.expires > t)
.map(|l| l.email.clone());
if let Some(email) = link {
self.commit(vec![del("link", &h)])?;
drop(_w);
self.audit(&email, "sign-in", "email link");
return self.session_for(&email);
}
let invite = self
.state()
.invites
.get(&h)
.filter(|i| i.expires > t)
.cloned();
if let Some(inv) = invite {
let mut changes = vec![del("invite", &h)];
if !self.state().users.contains_key(&inv.email) {
changes.push(put(
"user",
&inv.email,
&User {
role: inv.role,
created: t,
last_seen: None,
},
));
}
self.commit(changes)?;
drop(_w);
self.audit(&inv.email, "join", inv.role.as_str());
return self.session_for(&inv.email);
}
drop(_w);
self.fail(&from);
bail!("this link has expired or was already used; ask for a new one")
}
pub fn sign_in_code(
&self,
email: &str,
code: &str,
client: &str,
) -> Result<(String, Who, Option<String>)> {
let email = normalize(email);
anyhow::ensure!(valid_email(&email), "Wrong email or code");
let from = format!("code|{email}|{client}");
anyhow::ensure!(
self.failures(&from) < MAX_FAILED,
"Too many wrong codes for this address; try again in {LOCK_MINUTES} minutes"
);
let h = code_hash(code);
let t = now();
let _w = self.write_lock();
let invite = self
.state()
.invites
.get(&h)
.filter(|i| i.email == email && i.expires > t)
.cloned();
if let Some(inv) = invite {
let personal = new_code();
let mut changes = vec![
del("invite", &h),
put(
"code",
&email,
&Code {
hash: code_hash(&personal),
created: t.clone(),
},
),
];
if !self.state().users.contains_key(&email) {
changes.push(put(
"user",
&email,
&User {
role: inv.role,
created: t,
last_seen: None,
},
));
}
self.commit(changes)?;
drop(_w);
self.forgive(&from);
self.audit(&email, "join", inv.role.as_str());
let (session, who) = self.session_for(&email)?;
return Ok((session, who, Some(personal)));
}
drop(_w);
let known = self
.state()
.codes
.get(&email)
.is_some_and(|c| constant_eq(&c.hash, &h));
if known && self.member(&email)?.is_some() {
self.forgive(&from);
self.audit(&email, "sign-in", "code");
let (session, who) = self.session_for(&email)?;
return Ok((session, who, None));
}
self.fail(&from);
bail!("Wrong email or code")
}
pub fn reset_code(&self, by: &str, email: &str, keep: Option<&str>) -> Result<String> {
let email = normalize(email);
anyhow::ensure!(self.member(&email)?.is_some(), "{email} is not a member");
anyhow::ensure!(
by == email || !admin_emails().contains(&email),
"{email} is an admin by the deployment's settings and resets their own code"
);
let code = new_code();
let keep = keep.map(hash).unwrap_or_default();
let mut changes = vec![put(
"code",
&email,
&Code {
hash: code_hash(&code),
created: now(),
},
)];
changes.extend(
self.state()
.sessions
.iter()
.filter(|(k, s)| s.email == email && **k != keep)
.map(|(k, _)| del("session", k)),
);
self.commit(changes)?;
let prefix = format!("code|{email}|");
self.lock()
.failures
.retain(|(k, _)| !k.starts_with(&prefix));
self.audit(by, "reset-code", &email);
Ok(code)
}
fn session_for(&self, email: &str) -> Result<(String, Who)> {
let who = self.member(email)?.context("not a member")?;
let token = new_token("ses_");
let session = Session {
email: who.email.clone(),
created: now(),
expires: (Utc::now() + Duration::days(SESSION_DAYS)).to_rfc3339(),
};
self.commit(vec![put("session", &hash(&token), &session)])?;
Ok((token, who))
}
pub fn authenticate(
&self,
session: Option<&str>,
bearer: Option<&str>,
) -> Result<Option<(Who, Via)>> {
let t = now();
let mut found: Option<(String, Via)> = None;
let mut used: Option<(String, Token)> = None;
if let Some(s) = session {
found = self
.state()
.sessions
.get(&hash(s))
.filter(|s| s.expires > t)
.map(|s| (s.email.clone(), Via::Session));
}
if found.is_none() {
if let Some(b) = bearer {
if env_secret("COMINDLY_ADMIN_TOKEN").is_some_and(|a| constant_eq(b, &a)) {
found = admin_emails()
.into_iter()
.next()
.map(|e| (e, Via::AdminToken));
} else {
let h = hash(b);
let st = self.state();
if let Some((id, tok)) = st.tokens.iter().find(|(_, t)| t.hash == h) {
found = Some((tok.email.clone(), Via::Token));
used = Some((id.clone(), tok.clone()));
}
}
}
}
let Some((email, via)) = found else {
return Ok(None);
};
let who = self.member(&email)?;
if who.is_some() {
let since = (Utc::now() - Duration::minutes(5)).to_rfc3339();
let mut changes = Vec::new();
if let Some((id, mut tok)) =
used.filter(|(_, t)| t.last_used.as_ref().is_none_or(|u| *u < since))
{
tok.last_used = Some(t.clone());
changes.push(put("token", &id, &tok));
}
let (user, seen) = {
let st = self.state();
(st.users.get(&email).cloned(), st.seen.get(&email).cloned())
};
if let Some(mut u) = user {
if u.last_seen.as_ref().is_none_or(|s| *s < since) {
u.last_seen = Some(t);
changes.push(put("user", &email, &u));
}
} else if seen.is_none_or(|s| s < since) {
changes.push(put("seen", &email, &t));
}
if let Err(e) = self.commit(changes) {
eprintln!("comindly: recording last use: {e:#}");
}
}
Ok(who.map(|w| (w, via)))
}
pub fn sign_out(&self, session: &str, everywhere: bool) -> Result<()> {
let h = hash(session);
let changes = {
let st = self.state();
match (everywhere, st.sessions.get(&h)) {
(true, Some(s)) => st
.sessions
.iter()
.filter(|(_, x)| x.email == s.email)
.map(|(k, _)| del("session", k))
.collect(),
_ => vec![del("session", &h)],
}
};
self.commit(changes)
}
pub fn create_token(&self, who: &Who, name: &str) -> Result<String> {
let name = name.trim();
anyhow::ensure!(!name.is_empty(), "give the token a name");
let token = new_token("cly_");
let id = uuid::Uuid::new_v4().simple().to_string();
let row = Token {
hash: hash(&token),
email: who.email.clone(),
name: name.to_string(),
created: now(),
last_used: None,
};
self.commit(vec![put("token", &id, &row)])?;
self.audit(&who.email, "token", name);
Ok(token)
}
pub fn tokens(&self, who: &Who) -> Result<Vec<AccessToken>> {
let mut out: Vec<AccessToken> = self
.state()
.tokens
.iter()
.filter(|(_, t)| t.email == who.email)
.map(|(id, t)| AccessToken {
id: id.clone(),
name: t.name.clone(),
created: t.created.clone(),
last_used: t.last_used.clone(),
})
.collect();
out.sort_by(|a, b| b.created.cmp(&a.created));
Ok(out)
}
pub fn revoke_token(&self, who: &Who, id: &str) -> Result<()> {
let mine = self
.state()
.tokens
.get(id)
.is_some_and(|t| t.email == who.email);
if mine {
self.commit(vec![del("token", id)])?;
}
self.audit(&who.email, "revoke-token", id);
Ok(())
}
pub fn set_secret(&self, by: &str, name: &str, value: &str) -> Result<()> {
let value = value.trim();
if value.is_empty() {
self.commit(vec![del("secret", name)])?;
self.audit(by, "secret-removed", name);
return Ok(());
}
let key = env_secret("COMINDLY_SECRET_KEY")
.context("set COMINDLY_SECRET_KEY on the server to store tokens here")?;
let row = sealed(&key, name, value, by)?;
self.commit(vec![put("secret", name, &row)])?;
self.audit(by, "secret-set", name);
Ok(())
}
pub fn secret(&self, name: &str) -> Option<String> {
let key = env_secret("COMINDLY_SECRET_KEY")?;
let row = self.state().secrets.get(name).cloned()?;
unseal(&key, name, &row)
}
pub fn secret_info(&self, name: &str) -> Option<SecretInfo> {
self.state().secrets.get(name).map(|s| SecretInfo {
name: name.to_string(),
source: "stored",
hint: Some(s.hint.clone()),
updated: Some(s.updated.clone()),
updated_by: Some(s.updated_by.clone()),
})
}
pub fn secret_names(&self) -> Vec<String> {
self.state().secrets.keys().cloned().collect()
}
fn rotate_key(&self) -> Result<()> {
let (Some(key), Some(old)) = (
env_secret("COMINDLY_SECRET_KEY"),
env_secret("COMINDLY_SECRET_KEY_PREVIOUS"),
) else {
return Ok(());
};
let rows: Vec<(String, Secret)> = self
.state()
.secrets
.iter()
.map(|(n, s)| (n.clone(), s.clone()))
.collect();
let mut changes = Vec::new();
for (name, row) in rows {
if unseal(&key, &name, &row).is_some() {
continue;
}
let Some(plain) = unseal(&old, &name, &row) else {
continue;
};
let mut again = sealed(&key, &name, &plain, &row.updated_by)?;
again.updated = row.updated;
changes.push(put("secret", &name, &again));
}
self.commit(changes)
}
pub fn keep_config(&self, text: &str) -> Result<()> {
if self.state().config.get(CONFIG).map(String::as_str) == Some(text) {
return Ok(());
}
self.commit(vec![put("config", CONFIG, &text.to_string())])
}
pub fn kept_config(&self) -> Option<String> {
self.state().config.get(CONFIG).cloned()
}
fn write_lock(&self) -> std::sync::MutexGuard<'_, ()> {
self.writing.lock().unwrap_or_else(|e| e.into_inner())
}
}
fn sealed(key: &str, name: &str, value: &str, by: &str) -> Result<Secret> {
let (s, n) = seal(key, name, value)?;
Ok(Secret {
sealed: hex(&s),
nonce: hex(&n),
hint: hint(value),
updated: now(),
updated_by: by.to_string(),
})
}
fn unseal(key: &str, name: &str, row: &Secret) -> Option<String> {
open(key, name, &unhex(&row.sealed)?, &unhex(&row.nonce)?)
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
fn unhex(s: &str) -> Option<Vec<u8>> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(s.get(i..i + 2)?, 16).ok())
.collect()
}
fn constant_eq(a: &str, b: &str) -> bool {
let (a, b) = (blake3::hash(a.as_bytes()), blake3::hash(b.as_bytes()));
a.as_bytes()
.iter()
.zip(b.as_bytes())
.fold(0u8, |acc, (x, y)| acc | (x ^ y))
== 0
}
fn valid_email(email: &str) -> bool {
email.contains('@') && !email.contains(|c: char| c.is_whitespace() || c == '|')
}
static STORE: std::sync::OnceLock<Store> = std::sync::OnceLock::new();
pub fn init(index_root: &str) -> Result<&'static Store> {
if let Some(s) = STORE.get() {
return Ok(s);
}
let store = Store::open(index_root)?;
Ok(STORE.get_or_init(|| store))
}
pub fn store() -> Option<&'static Store> {
static ATTACHED: std::sync::OnceLock<Option<Store>> = std::sync::OnceLock::new();
STORE.get().or_else(|| {
ATTACHED
.get_or_init(|| {
let key = env_secret("COMINDLY_SECRET_KEY").filter(|_| enabled())?;
let root = crate::config::Config::load().index_dir(None);
let previous = env_secret("COMINDLY_SECRET_KEY_PREVIOUS");
Store::attach_keyed(&root, &key, previous.as_deref())
.map_err(|e| eprintln!("comindly: the server's stored keys: {e:#}"))
.ok()
})
.as_ref()
})
}
pub fn git_secret(host: &str) -> String {
format!("git:{}", host.to_lowercase())
}
pub const LLM_SECRET: &str = "llm_api_key";
pub fn git_token(host: &str) -> Option<String> {
store()?.secret(&git_secret(host))
}
pub fn mail_configured() -> bool {
env_secret("COMINDLY_SMTP_URL").is_some()
}
pub async fn send_mail(to: &str, subject: &str, body: &str) -> Result<()> {
use lettre::message::header::ContentType;
use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};
let url = env_secret("COMINDLY_SMTP_URL").context("no mail server configured")?;
let from =
env_secret("COMINDLY_SMTP_FROM").unwrap_or_else(|| "comindly <comindly@localhost>".into());
let msg = Message::builder()
.from(
from.parse()
.context("COMINDLY_SMTP_FROM is not an address")?,
)
.to(to.parse().context("not an email address")?)
.subject(subject)
.header(ContentType::TEXT_PLAIN)
.body(body.to_string())?;
AsyncSmtpTransport::<Tokio1Executor>::from_url(&url)
.context("COMINDLY_SMTP_URL is not a valid smtp:// or smtps:// URL")?
.build()
.send(msg)
.await
.map_err(|e| anyhow::anyhow!("sending mail failed: {e}"))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_store() -> ((), Store) {
let dir = std::env::temp_dir().join(format!("comindly-server-{}", uuid::Uuid::new_v4()));
(
(),
Store::open_keyed(dir.to_str().unwrap(), "test key", None).unwrap(),
)
}
#[test]
fn invitations_are_single_use_and_tied_to_their_email() {
let (_d, s) = temp_store();
let token = s
.invite("admin@x.org", "Dev@X.org ", Role::Member, false)
.unwrap();
let (session, who) = s.redeem(&token, "10.0.0.1").unwrap();
assert_eq!(who.email, "dev@x.org");
assert_eq!(who.role, Role::Member);
assert!(s.redeem(&token, "10.0.0.1").is_err());
let (back, _) = s.authenticate(Some(&session), None).unwrap().unwrap();
assert_eq!(back.email, "dev@x.org");
s.sign_out(&session, false).unwrap();
assert!(s.authenticate(Some(&session), None).unwrap().is_none());
}
#[test]
fn sign_in_links_only_for_members_and_once() {
let (_d, s) = temp_store();
assert!(s.sign_in_link("stranger@x.org").unwrap().is_none());
let inv = s.invite("a", "m@x.org", Role::Member, false).unwrap();
s.redeem(&inv, "10.0.0.1").unwrap();
let link = s.sign_in_link("m@x.org").unwrap().unwrap();
assert!(s.redeem(&link, "10.0.0.1").is_ok());
assert!(s.redeem(&link, "10.0.0.1").is_err());
}
#[test]
fn removing_a_member_ends_their_tokens() {
let (_d, s) = temp_store();
let inv = s.invite("a", "m@x.org", Role::Member, false).unwrap();
let (_, who) = s.redeem(&inv, "10.0.0.1").unwrap();
let token = s.create_token(&who, "laptop").unwrap();
assert!(s.authenticate(None, Some(&token)).unwrap().is_some());
s.remove("a", "m@x.org").unwrap();
assert!(s.authenticate(None, Some(&token)).unwrap().is_none());
}
#[test]
fn invitation_codes_need_their_own_address_and_give_a_sign_in_code() {
let (_d, s) = temp_store();
let invite = s.invite("a", "m@x.org", Role::Member, true).unwrap();
assert!(s.sign_in_code("other@x.org", &invite, "10.0.0.1").is_err());
let typed = invite.to_lowercase().replace('-', " ");
let (_, who, personal) = s.sign_in_code("M@x.org", &typed, "10.0.0.1").unwrap();
assert_eq!(who.email, "m@x.org");
let personal = personal.unwrap();
assert!(s.sign_in_code("m@x.org", &invite, "10.0.0.1").is_err());
let (_, _, again) = s.sign_in_code("m@x.org", &personal, "10.0.0.1").unwrap();
assert!(again.is_none());
}
#[test]
fn a_reset_code_replaces_the_old_one_and_ends_other_sessions() {
let (_d, s) = temp_store();
let invite = s.invite("a", "m@x.org", Role::Member, true).unwrap();
let (mine, _, old) = s.sign_in_code("m@x.org", &invite, "10.0.0.1").unwrap();
let (elsewhere, _, _) = s
.sign_in_code("m@x.org", old.as_ref().unwrap(), "10.0.0.1")
.unwrap();
let new = s.reset_code("m@x.org", "m@x.org", Some(&mine)).unwrap();
assert!(s.authenticate(Some(&mine), None).unwrap().is_some());
assert!(s.authenticate(Some(&elsewhere), None).unwrap().is_none());
assert!(s
.sign_in_code("m@x.org", old.as_ref().unwrap(), "10.0.0.1")
.is_err());
assert!(s.sign_in_code("m@x.org", &new, "10.0.0.1").is_ok());
}
#[test]
fn wrong_codes_lock_the_address_for_that_client() {
let (_d, s) = temp_store();
let invite = s.invite("a", "m@x.org", Role::Member, true).unwrap();
for _ in 0..MAX_FAILED {
assert!(s.sign_in_code("m@x.org", "AAAA-AAAA", "10.0.0.1").is_err());
}
let e = s
.sign_in_code("m@x.org", &invite, "10.0.0.1")
.unwrap_err()
.to_string();
assert!(e.contains("Too many"), "{e}");
assert!(s.sign_in_code("m@x.org", &invite, "10.0.0.2").is_ok());
}
#[test]
fn accounts_survive_a_restart_and_rows_without_the_key_are_ignored() {
let dir = std::env::temp_dir().join(format!("comindly-server-{}", uuid::Uuid::new_v4()));
let root = dir.to_str().unwrap();
let s = Store::open_keyed(root, "key one", None).unwrap();
let code = s
.invite("admin@x.org", "dev@x.org", Role::Member, true)
.unwrap();
let (_, _, personal) = s.sign_in_code("dev@x.org", &code, "1.2.3.4").unwrap();
let token = s
.create_token(&s.member("dev@x.org").unwrap().unwrap(), "agent")
.unwrap();
drop(s);
let again = Store::open_keyed(root, "key one", None).unwrap();
assert!(again
.sign_in_code("dev@x.org", &personal.unwrap(), "1.2.3.4")
.is_ok());
assert!(again.authenticate(None, Some(&token)).unwrap().is_some());
let forged = vec![put(
"user",
"eve@x.org",
&User {
role: Role::Admin,
created: now(),
last_seen: None,
},
)];
run(write(
again.uri.clone(),
mac_key("someone else's key"),
forged,
))
.unwrap();
again.reload().unwrap();
assert!(again.member("eve@x.org").unwrap().is_none());
let rotated = Store::open_keyed(root, "key two", Some("key one")).unwrap();
assert!(rotated.authenticate(None, Some(&token)).unwrap().is_some());
let only_new = Store::open_keyed(root, "key two", None).unwrap();
assert!(only_new.authenticate(None, Some(&token)).unwrap().is_some());
}
#[test]
fn codes_are_twenty_characters_in_groups() {
let c = new_code();
assert_eq!(c.len(), 24);
assert!(c.split('-').all(|g| g.len() == 4));
assert_ne!(c, new_code());
}
#[test]
fn sealed_values_open_only_with_their_key() {
let (sealed, nonce) = seal("key-one", "git:gitlab.com", "glpat-example").unwrap();
assert_eq!(
open("key-one", "git:gitlab.com", &sealed, &nonce).as_deref(),
Some("glpat-example")
);
assert!(open("key-two", "git:gitlab.com", &sealed, &nonce).is_none());
assert!(open("key-one", "llm_api_key", &sealed, &nonce).is_none());
assert_eq!(hint("glpat-example"), "…mple");
}
}