diff --git a/migrations/067_team_policy.sql b/migrations/067_team_policy.sql new file mode 100644 index 0000000..520d1d1 --- /dev/null +++ b/migrations/067_team_policy.sql @@ -0,0 +1,7 @@ +ALTER TABLE teams + ADD COLUMN IF NOT EXISTS policy JSONB NULL, + ADD COLUMN IF NOT EXISTS policy_revision BIGINT NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS policy_updated_at TIMESTAMPTZ NULL, + ADD COLUMN IF NOT EXISTS policy_updated_by UUID NULL REFERENCES users(id) ON DELETE SET NULL; + +UPDATE teams SET policy = jsonb_build_object('version', 1, 'policy', jsonb_build_object('security.lockTimeoutMinutes', jsonb_build_object('$range', jsonb_build_object('max', lock_max_minutes))) || CASE WHEN lock_force_vault THEN jsonb_build_object('security.lockAction', 'vault') ELSE '{}'::jsonb END), policy_revision = 1 WHERE lock_max_minutes IS NOT NULL AND policy IS NULL; diff --git a/src/db.rs b/src/db.rs index 68755f8..7b4ed73 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,7 +1,15 @@ +use sqlx::migrate::Migrator; use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; use tracing::{error, info}; +/// Ignores applied versions this build lacks, so a rollback to an older image still boots. +pub(crate) fn migrator() -> Migrator { + let mut migrator = sqlx::migrate!("./migrations"); + migrator.set_ignore_missing(true); + migrator +} + pub async fn create_pool() -> PgPool { let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set"); @@ -18,7 +26,7 @@ pub async fn create_pool() -> PgPool { info!("PostgreSQL connection pool is ready"); - sqlx::migrate!("./migrations") + migrator() .run(&pool) .await .unwrap_or_else(|err| { @@ -28,10 +36,21 @@ pub async fn create_pool() -> PgPool { info!("Migrations applied successfully"); reconcile_legacy_grants(&pool).await; + backfill_team_policies(&pool).await; pool } +pub(crate) async fn backfill_team_policies(pool: &PgPool) { + match sqlx::query(crate::team_policy::LOCK_POLICY_BACKFILL) + .execute(pool) + .await + { + Ok(res) => info!(filled = res.rows_affected(), "Backfilled team policy documents"), + Err(err) => error!(error = %err, "Failed to backfill team policy documents"), + } +} + /// Migration 037's backfill is one-shot: a rollback-then-forward-again cycle /// can leave live invite_link sessions with an `invite_token` but no grant, /// since nothing reads that column outside the grants path any more. Runs @@ -120,4 +139,28 @@ mod tests { .unwrap(); assert_eq!(count, 1, "second run must insert nothing new"); } + + #[tokio::test] + async fn migrator_boots_past_versions_this_build_lacks() { + let pool = test_pool_or_skip!(); + let mut tx = pool.begin().await.unwrap(); + sqlx::query( + "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) \ + VALUES (9000000000000000000, 'from a newer image', TRUE, '\\x00'::bytea, 0)", + ) + .execute(&mut *tx) + .await + .unwrap(); + + let mut strict = sqlx::migrate!("./migrations"); + strict.set_locking(false); + assert!(matches!( + strict.run(&mut *tx).await, + Err(sqlx::migrate::MigrateError::VersionMissing(9000000000000000000)) + )); + let mut lenient = migrator(); + lenient.set_locking(false); + lenient.run(&mut *tx).await.unwrap(); + tx.rollback().await.unwrap(); + } } diff --git a/src/main.rs b/src/main.rs index 5fc38f0..76243e9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,6 +18,7 @@ mod session_grants; mod team_authority; mod team_join_grants; mod team_plan; +mod team_policy; mod vault_key_exposure; mod sync_notifier; mod terminal_manager; @@ -456,6 +457,12 @@ async fn main() { "/v1/teams/:team_id/lock-policy", put(routes::team_lock_policy::set_lock_policy).delete(routes::team_lock_policy::clear_lock_policy), ) + .route( + "/v1/teams/:team_id/policy", + get(routes::team_policy::get_policy) + .put(routes::team_policy::put_policy) + .delete(routes::team_policy::delete_policy), + ) .route( "/v1/teams/:team_id/members/:user_id", delete(routes::teams::remove_member), diff --git a/src/routes/mod.rs b/src/routes/mod.rs index fac33d7..cb31d2b 100644 --- a/src/routes/mod.rs +++ b/src/routes/mod.rs @@ -16,6 +16,7 @@ pub mod session_codes; pub mod sync; pub mod team_grants; pub mod team_lock_policy; +pub mod team_policy; pub mod team_sync; pub mod team_objects; pub mod team_object_prefs; diff --git a/src/routes/team_lock_policy.rs b/src/routes/team_lock_policy.rs index 2ddf607..a50681c 100644 --- a/src/routes/team_lock_policy.rs +++ b/src/routes/team_lock_policy.rs @@ -3,34 +3,16 @@ use axum::{ http::StatusCode, Extension, Json, }; -use serde::{Deserialize, Serialize}; use serde_json::json; use sqlx::PgPool; -use tracing::error; use uuid::Uuid; use crate::auth::AuthUser; -use crate::routes::audit::write_audit_event; -use crate::routes::teams::{notify_team_members_changed, require_vault_manager}; +use crate::routes::team_policy::{announce, LockedPolicy}; +use crate::routes::teams::require_vault_manager; use crate::sync_notifier::SyncNotifier; use crate::team_plan::require_granular; - -const LOCK_MINUTES: [i32; 6] = [0, 5, 15, 30, 60, 240]; - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub struct LockPolicy { - pub max_minutes: i32, - pub force_vault: bool, -} - -impl LockPolicy { - pub fn from_columns(max_minutes: Option, force_vault: bool) -> Option { - max_minutes.map(|max_minutes| Self { - max_minutes, - force_vault, - }) - } -} +use crate::team_policy::{with_lock_policy, LockPolicy, LOCK_MINUTES}; pub async fn set_lock_policy( State(pool): State, @@ -65,54 +47,28 @@ async fn write_policy( actor: Uuid, policy: Option, ) -> Result { - sqlx::query("UPDATE teams SET lock_max_minutes = $2, lock_force_vault = $3 WHERE id = $1") - .bind(team_id) - .bind(policy.map(|p| p.max_minutes)) - .bind(policy.is_some_and(|p| p.force_vault)) - .execute(pool) - .await - .map_err(|e| { - error!(error = %e, team_id = %team_id, "Failed to write team lock policy"); - StatusCode::INTERNAL_SERVER_ERROR - })?; + let mut locked = LockedPolicy::lock(pool, team_id).await?; + let doc = with_lock_policy(locked.doc.take(), policy); + locked.replace(actor, doc.as_ref()).await?; let (action, metadata) = match policy { Some(p) => ("team.lock_policy_set", Some(json!(p))), None => ("team.lock_policy_removed", None), }; - write_audit_event( - pool.clone(), - team_id, - actor, - action, - Some("team"), - Some(team_id.to_string()), - None, - metadata, - ) - .await; - notify_team_members_changed(pool, notifier, team_id).await; + announce(pool, notifier, team_id, actor, action, metadata).await; Ok(StatusCode::NO_CONTENT) } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use crate::permissions::{PERM_MANAGE_MEMBERS, PERM_MANAGE_VAULT}; - use crate::routes::teams::list_teams; - use crate::sync_notifier::SyncEvent; use crate::test_support::{ - member_with_role, seed_team_with_roles, seed_user, set_user_tier, BillingMode, + listed_team, member_with_role, seed_team_with_roles, seed_user, set_user_tier, + team_audit_count as audit_count, team_members_told, BillingMode, }; async fn listed(pool: &PgPool, team: Uuid, user: Uuid) -> Option { - let Json(teams) = list_teams(State(pool.clone()), Extension(AuthUser(user))) - .await - .unwrap(); - teams - .into_iter() - .find(|t| t.id == team) - .unwrap() - .lock_policy + listed_team(pool, team, user).await.lock_policy } #[tokio::test] @@ -139,7 +95,7 @@ mod tests { ); } - async fn put( + pub(crate) async fn put( pool: &PgPool, team: Uuid, user: Uuid, @@ -159,7 +115,7 @@ mod tests { .await } - async fn clear(pool: &PgPool, team: Uuid, user: Uuid) -> Result { + pub(crate) async fn clear(pool: &PgPool, team: Uuid, user: Uuid) -> Result { clear_lock_policy( State(pool.clone()), Extension(AuthUser(user)), @@ -169,15 +125,6 @@ mod tests { .await } - async fn audit_count(pool: &PgPool, team: Uuid, action: &str) -> i64 { - sqlx::query_scalar("SELECT count(*) FROM audit_logs WHERE team_id = $1 AND action = $2") - .bind(team) - .bind(action) - .fetch_one(pool) - .await - .unwrap() - } - #[tokio::test] async fn owner_sets_and_clears_the_policy_every_member_lists() { let _env = BillingMode::self_hosted(); @@ -328,15 +275,7 @@ mod tests { .await .unwrap(); - let want = format!("team_members:{team}"); - let mut told = std::collections::HashSet::new(); - while let Ok(ev) = rx.try_recv() { - if let SyncEvent::BlobPushed { user_id, device_id } = ev { - if device_id == want { - told.insert(user_id); - } - } - } + let told = team_members_told(&mut rx, team); assert!(told.contains(&owner) && told.contains(&member)); } } diff --git a/src/routes/team_policy.rs b/src/routes/team_policy.rs new file mode 100644 index 0000000..1372175 --- /dev/null +++ b/src/routes/team_policy.rs @@ -0,0 +1,622 @@ +use axum::{ + extract::{Path, State}, + http::StatusCode, + response::{IntoResponse, Response}, + Extension, Json, +}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sqlx::{PgPool, Postgres, Transaction}; +use tracing::error; +use uuid::Uuid; + +use crate::auth::AuthUser; +use crate::permissions::require_team_member; +use crate::routes::audit::write_audit_event; +use crate::routes::coded_error; +use crate::routes::teams::{notify_team_members_changed, require_vault_manager}; +use crate::sync_notifier::SyncNotifier; +use crate::team_plan::team_locked; +use crate::team_policy::{is_narrowing, key_diff, lock_policy_of, validate_doc, DocError}; + +#[derive(Serialize, sqlx::FromRow)] +pub struct TeamPolicy { + policy: Option, + revision: i64, + updated_at: Option>, + updated_by: Option, +} + +#[derive(Deserialize)] +pub struct PutPolicy { + pub policy: Value, + pub revision: i64, +} + +fn db_error(e: sqlx::Error) -> StatusCode { + error!(error = %e, "Team policy query failed"); + StatusCode::INTERNAL_SERVER_ERROR +} + +pub enum PolicyRefusal { + Doc(DocError), + RevisionMismatch, + Status(StatusCode), +} + +impl From for PolicyRefusal { + fn from(status: StatusCode) -> Self { + Self::Status(status) + } +} + +impl IntoResponse for PolicyRefusal { + fn into_response(self) -> Response { + match self { + Self::Doc(e) => { + let mut body = json!({ "error": e.code() }); + if let Some(key) = e.key() { + body["key"] = json!(key); + } + (StatusCode::BAD_REQUEST, Json(body)).into_response() + } + Self::RevisionMismatch => coded_error(StatusCode::CONFLICT, "POLICY_REVISION_MISMATCH"), + Self::Status(status) => status.into_response(), + } + } +} + +pub(crate) struct LockedPolicy { + tx: Transaction<'static, Postgres>, + team_id: Uuid, + pub doc: Option, + pub revision: i64, +} + +impl LockedPolicy { + pub(crate) async fn lock(pool: &PgPool, team_id: Uuid) -> Result { + let mut tx = pool.begin().await.map_err(db_error)?; + let (doc, revision) = + sqlx::query_as("SELECT policy, policy_revision FROM teams WHERE id = $1 FOR UPDATE") + .bind(team_id) + .fetch_optional(&mut *tx) + .await + .map_err(db_error)? + .ok_or(StatusCode::NOT_FOUND)?; + Ok(Self { + tx, + team_id, + doc, + revision, + }) + } + + /// Also mirrors the app-lock keys into the `lock_*` columns so a server rollback keeps them. + pub(crate) async fn replace( + mut self, + actor: Uuid, + doc: Option<&Value>, + ) -> Result { + let lock = lock_policy_of(doc); + let revision = sqlx::query_scalar( + "UPDATE teams SET policy = $2, policy_revision = policy_revision + 1, + policy_updated_at = now(), policy_updated_by = $3, + lock_max_minutes = $4, lock_force_vault = $5 + WHERE id = $1 RETURNING policy_revision", + ) + .bind(self.team_id) + .bind(doc) + .bind(actor) + .bind(lock.map(|p| p.max_minutes)) + .bind(lock.is_some_and(|p| p.force_vault)) + .fetch_one(&mut *self.tx) + .await + .map_err(db_error)?; + self.tx.commit().await.map_err(db_error)?; + Ok(revision) + } +} + +pub(crate) async fn announce( + pool: &PgPool, + notifier: &SyncNotifier, + team_id: Uuid, + actor: Uuid, + action: &str, + metadata: Option, +) { + write_audit_event( + pool.clone(), + team_id, + actor, + action, + Some("team"), + Some(team_id.to_string()), + None, + metadata, + ) + .await; + notify_team_members_changed(pool, notifier, team_id).await; +} + +pub async fn get_policy( + State(pool): State, + Extension(auth): Extension, + Path(team_id): Path, +) -> Result, StatusCode> { + require_team_member(&pool, team_id, auth.0).await?; + sqlx::query_as( + "SELECT policy, policy_revision AS revision, policy_updated_at AS updated_at, + policy_updated_by AS updated_by + FROM teams WHERE id = $1", + ) + .bind(team_id) + .fetch_one(&pool) + .await + .map(Json) + .map_err(db_error) +} + +pub async fn put_policy( + State(pool): State, + Extension(auth): Extension, + Extension(notifier): Extension, + Path(team_id): Path, + Json(body): Json, +) -> Result { + require_vault_manager(&pool, team_id, auth.0).await?; + validate_doc(&body.policy).map_err(PolicyRefusal::Doc)?; + let plan_locked = team_locked(&pool, team_id).await?; + let locked = LockedPolicy::lock(&pool, team_id).await?; + if locked.revision != body.revision { + return Err(PolicyRefusal::RevisionMismatch); + } + if plan_locked && !is_narrowing(locked.doc.as_ref(), &body.policy) { + return Err(StatusCode::PAYMENT_REQUIRED.into()); + } + let diff = key_diff(locked.doc.as_ref(), Some(&body.policy)); + let revision = locked.replace(auth.0, Some(&body.policy)).await?; + let metadata = json!({ "diff": diff, "revision": revision }); + announce( + &pool, + ¬ifier, + team_id, + auth.0, + "team.policy_set", + Some(metadata), + ) + .await; + Ok(StatusCode::NO_CONTENT) +} + +pub async fn delete_policy( + State(pool): State, + Extension(auth): Extension, + Extension(notifier): Extension, + Path(team_id): Path, +) -> Result { + require_vault_manager(&pool, team_id, auth.0).await?; + let locked = LockedPolicy::lock(&pool, team_id).await?; + if locked.doc.is_none() { + return Ok(StatusCode::NO_CONTENT); + } + locked.replace(auth.0, None).await?; + announce( + &pool, + ¬ifier, + team_id, + auth.0, + "team.policy_removed", + None, + ) + .await; + Ok(StatusCode::NO_CONTENT) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::permissions::{PERM_MANAGE_MEMBERS, PERM_MANAGE_VAULT}; + use crate::routes::team_lock_policy::tests::{clear as clear_lock, put as put_lock}; + use crate::team_policy::LockPolicy; + use crate::test_support::{ + listed_team as listed, member_with_role, seed_team_with_roles, seed_user, set_user_tier, + team_audit_count as audit_count, team_members_told, BillingMode, + }; + + fn doc(policy: Value) -> Value { + json!({"version": 1, "policy": policy}) + } + + async fn put( + pool: &PgPool, + notifier: &SyncNotifier, + team: Uuid, + user: Uuid, + policy: Value, + revision: i64, + ) -> (StatusCode, Value) { + let resp = put_policy( + State(pool.clone()), + Extension(AuthUser(user)), + Extension(notifier.clone()), + Path(team), + Json(PutPolicy { policy, revision }), + ) + .await + .into_response(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) + } + + async fn put_status( + pool: &PgPool, + team: Uuid, + user: Uuid, + policy: Value, + revision: i64, + ) -> StatusCode { + put(pool, &SyncNotifier::new(), team, user, policy, revision) + .await + .0 + } + + async fn delete( + pool: &PgPool, + notifier: &SyncNotifier, + team: Uuid, + user: Uuid, + ) -> Result { + delete_policy( + State(pool.clone()), + Extension(AuthUser(user)), + Extension(notifier.clone()), + Path(team), + ) + .await + } + + async fn get(pool: &PgPool, team: Uuid, user: Uuid) -> Result { + get_policy(State(pool.clone()), Extension(AuthUser(user)), Path(team)) + .await + .map(|Json(r)| serde_json::to_value(r).unwrap()) + } + + async fn lock_columns(pool: &PgPool, team: Uuid) -> (Option, bool) { + sqlx::query_as("SELECT lock_max_minutes, lock_force_vault FROM teams WHERE id = $1") + .bind(team) + .fetch_one(pool) + .await + .unwrap() + } + + async fn self_hosted_team() -> Option<(BillingMode, PgPool, Uuid, Uuid)> { + let env = BillingMode::self_hosted(); + let pool = crate::test_support::test_pool().await?; + let owner = seed_user(&pool).await; + let team = seed_team_with_roles(&pool, owner).await; + Some((env, pool, owner, team)) + } + + #[tokio::test] + async fn put_then_list_teams_returns_doc_revision_and_joined_at() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let policy = doc(json!({"sync.types.key": false})); + + assert_eq!( + put_status(&pool, team, owner, policy.clone(), 0).await, + StatusCode::NO_CONTENT + ); + + let joined_at: chrono::DateTime = sqlx::query_scalar( + "SELECT joined_at FROM team_members WHERE team_id = $1 AND user_id = $2", + ) + .bind(team) + .bind(owner) + .fetch_one(&pool) + .await + .unwrap(); + let t = listed(&pool, team, owner).await; + assert_eq!(t.policy, Some(policy.clone())); + assert_eq!(t.policy_revision, 1); + assert_eq!(t.joined_at, joined_at); + let json = serde_json::to_value(&t).unwrap(); + assert_eq!(json["policy"], policy); + assert_eq!(json["policy_revision"], 1); + assert!(json["joined_at"].is_string()); + + let got = get(&pool, team, owner).await.unwrap(); + assert_eq!(got["policy"], policy); + assert_eq!(got["revision"], 1); + assert_eq!(got["updated_by"], json!(owner)); + assert!(got["updated_at"].is_string()); + } + + #[tokio::test] + async fn put_with_stale_revision_is_409_and_keeps_stored() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let first = doc(json!({"a.b": 1})); + assert_eq!( + put_status(&pool, team, owner, first.clone(), 0).await, + StatusCode::NO_CONTENT + ); + + let (status, body) = put( + &pool, + &SyncNotifier::new(), + team, + owner, + doc(json!({"a.b": 2})), + 0, + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + assert_eq!(body, json!({"error": "POLICY_REVISION_MISMATCH"})); + + let got = get(&pool, team, owner).await.unwrap(); + assert_eq!(got["policy"], first); + assert_eq!(got["revision"], 1); + } + + #[tokio::test] + async fn put_rejects_machine_only_key_with_key_in_body() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let (status, body) = put( + &pool, + &SyncNotifier::new(), + team, + owner, + doc(json!({"server.url": "https://example.com"})), + 0, + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!( + body, + json!({"error": "POLICY_MACHINE_ONLY_KEY", "key": "server.url"}) + ); + + let (status, body) = put( + &pool, + &SyncNotifier::new(), + team, + owner, + json!({"version": 2}), + 0, + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body, json!({"error": "POLICY_INVALID"})); + assert_eq!( + get(&pool, team, owner).await.unwrap()["policy"], + Value::Null + ); + } + + #[tokio::test] + async fn member_without_manage_vault_gets_403_on_put_and_200_on_get() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let member = member_with_role(&pool, team, PERM_MANAGE_MEMBERS).await; + let manager = member_with_role(&pool, team, PERM_MANAGE_VAULT).await; + let outsider = seed_user(&pool).await; + let policy = doc(json!({"a.b": 1})); + + assert_eq!( + put_status(&pool, team, member, policy.clone(), 0).await, + StatusCode::FORBIDDEN + ); + assert_eq!( + delete(&pool, &SyncNotifier::new(), team, member).await, + Err(StatusCode::FORBIDDEN) + ); + assert_eq!( + put_status(&pool, team, manager, policy.clone(), 0).await, + StatusCode::NO_CONTENT + ); + assert_eq!(get(&pool, team, member).await.unwrap()["policy"], policy); + assert_eq!(get(&pool, team, outsider).await, Err(StatusCode::FORBIDDEN)); + assert_eq!( + get(&pool, team, owner).await.unwrap()["updated_by"], + json!(manager) + ); + } + + #[tokio::test] + async fn below_business_put_402_but_removal_put_and_delete_allowed() { + let _env = BillingMode::hosted(); + let Some(pool) = crate::test_support::test_pool().await else { + return; + }; + let owner = seed_user(&pool).await; + set_user_tier(&pool, owner, "business").await; + let team = seed_team_with_roles(&pool, owner).await; + assert_eq!( + put_status(&pool, team, owner, doc(json!({"a.b": 1, "c.d": 2})), 0).await, + StatusCode::NO_CONTENT + ); + set_user_tier(&pool, owner, "teams").await; + + assert_eq!( + put_status( + &pool, + team, + owner, + doc(json!({"a.b": 1, "c.d": 2, "e.f": 3})), + 1 + ) + .await, + StatusCode::PAYMENT_REQUIRED + ); + assert_eq!( + put_status(&pool, team, owner, doc(json!({"a.b": 1})), 1).await, + StatusCode::NO_CONTENT + ); + assert_eq!( + put_status(&pool, team, owner, doc(json!({"a.b": 9})), 2).await, + StatusCode::PAYMENT_REQUIRED + ); + assert_eq!( + get(&pool, team, owner).await.unwrap()["policy"], + doc(json!({"a.b": 1})) + ); + assert_eq!( + delete(&pool, &SyncNotifier::new(), team, owner).await, + Ok(StatusCode::NO_CONTENT) + ); + assert_eq!( + get(&pool, team, owner).await.unwrap()["policy"], + Value::Null + ); + } + + #[tokio::test] + async fn put_and_delete_write_audit_and_notify_every_member() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let member = member_with_role(&pool, team, PERM_MANAGE_MEMBERS).await; + let notifier = SyncNotifier::new(); + let mut rx = notifier.subscribe(); + + let (status, _) = put(&pool, ¬ifier, team, owner, doc(json!({"a.b": 1})), 0).await; + assert_eq!(status, StatusCode::NO_CONTENT); + assert_eq!(audit_count(&pool, team, "team.policy_set").await, 1); + let metadata: Value = sqlx::query_scalar( + "SELECT metadata FROM audit_logs WHERE team_id = $1 AND action = 'team.policy_set'", + ) + .bind(team) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + metadata, + json!({"diff": {"added": ["a.b"], "changed": [], "removed": []}, "revision": 1}) + ); + let told = team_members_told(&mut rx, team); + assert!(told.contains(&owner) && told.contains(&member)); + + assert_eq!( + delete(&pool, ¬ifier, team, owner).await, + Ok(StatusCode::NO_CONTENT) + ); + assert_eq!(audit_count(&pool, team, "team.policy_removed").await, 1); + let told = team_members_told(&mut rx, team); + assert!(told.contains(&owner) && told.contains(&member)); + assert_eq!(get(&pool, team, owner).await.unwrap()["revision"], 2); + + assert_eq!( + delete(&pool, ¬ifier, team, owner).await, + Ok(StatusCode::NO_CONTENT) + ); + assert_eq!(audit_count(&pool, team, "team.policy_removed").await, 1); + assert!(team_members_told(&mut rx, team).is_empty()); + assert_eq!(get(&pool, team, owner).await.unwrap()["revision"], 2); + } + + #[tokio::test] + async fn lock_policy_put_writes_doc_and_keeps_other_keys() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + assert_eq!( + put_status(&pool, team, owner, doc(json!({"sync.types.key": false})), 0).await, + StatusCode::NO_CONTENT + ); + + assert_eq!( + put_lock(&pool, team, owner, 15, true).await, + Ok(StatusCode::NO_CONTENT) + ); + + let t = listed(&pool, team, owner).await; + assert_eq!( + t.policy, + Some(doc(json!({ + "sync.types.key": false, + "security.lockTimeoutMinutes": {"$range": {"max": 15}}, + "security.lockAction": "vault" + }))) + ); + assert_eq!(t.policy_revision, 2); + assert_eq!( + t.lock_policy, + Some(LockPolicy { + max_minutes: 15, + force_vault: true + }) + ); + assert_eq!(lock_columns(&pool, team).await, (Some(15), true)); + } + + #[tokio::test] + async fn lock_policy_delete_removes_only_app_lock_keys() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let other = doc(json!({"sync.types.key": false})); + assert_eq!( + put_status(&pool, team, owner, other.clone(), 0).await, + StatusCode::NO_CONTENT + ); + assert_eq!( + put_lock(&pool, team, owner, 30, true).await, + Ok(StatusCode::NO_CONTENT) + ); + assert_eq!( + clear_lock(&pool, team, owner).await, + Ok(StatusCode::NO_CONTENT) + ); + + let t = listed(&pool, team, owner).await; + assert_eq!(t.policy, Some(other)); + assert_eq!(t.policy_revision, 3); + assert_eq!(t.lock_policy, None); + assert_eq!(lock_columns(&pool, team).await, (None, false)); + } + + #[tokio::test] + async fn teams_list_derives_lock_policy_from_doc() { + let Some((_env, pool, owner, team)) = self_hosted_team().await else { + return; + }; + let policy = doc(json!({"security.lockTimeoutMinutes": {"$range": {"max": 20}}})); + sqlx::query( + "UPDATE teams SET policy = $2, lock_max_minutes = 60, lock_force_vault = TRUE WHERE id = $1", + ) + .bind(team) + .bind(&policy) + .execute(&pool) + .await + .unwrap(); + + let t = listed(&pool, team, owner).await; + assert_eq!( + t.lock_policy, + Some(LockPolicy { + max_minutes: 15, + force_vault: false + }) + ); + + assert_eq!( + put_status(&pool, team, owner, policy, 0).await, + StatusCode::NO_CONTENT + ); + assert_eq!(lock_columns(&pool, team).await, (Some(15), false)); + } +} diff --git a/src/routes/teams.rs b/src/routes/teams.rs index 52a2d84..5f85535 100644 --- a/src/routes/teams.rs +++ b/src/routes/teams.rs @@ -192,7 +192,10 @@ pub struct TeamWithRole { pub role_ids: Vec, pub permission_allow: i64, pub permission_deny: i64, - pub lock_policy: Option, + pub lock_policy: Option, + pub policy: Option, + pub policy_revision: i64, + pub joined_at: chrono::DateTime, } pub async fn list_teams( @@ -225,6 +228,9 @@ async fn teams_for_user( i64, Option, bool, + Option, + i64, + chrono::DateTime, String, Option>, bool, @@ -234,7 +240,7 @@ async fn teams_for_user( r#" SELECT t.id, t.name, t.owner_id, t.created_at, tmr.role_id, COALESCE(o.allow_mask, 0), COALESCE(o.deny_mask, 0), - t.lock_max_minutes, t.lock_force_vault, + t.lock_max_minutes, t.lock_force_vault, t.policy, t.policy_revision, tm.joined_at, {} FROM teams t JOIN team_members tm ON tm.team_id = t.id AND tm.user_id = $1 @@ -262,6 +268,9 @@ async fn teams_for_user( permission_deny, lock_max_minutes, lock_force_vault, + policy, + policy_revision, + joined_at, tier, trial_ends_at, admin_override, @@ -286,7 +295,13 @@ async fn teams_for_user( role_ids: role_id.into_iter().collect(), permission_allow, permission_deny, - lock_policy: crate::routes::team_lock_policy::LockPolicy::from_columns(lock_max_minutes, lock_force_vault), + lock_policy: match &policy { + Some(doc) => crate::team_policy::lock_policy_of(Some(doc)), + None => crate::team_policy::LockPolicy::from_columns(lock_max_minutes, lock_force_vault), + }, + policy, + policy_revision, + joined_at, }); } } diff --git a/src/team_policy.rs b/src/team_policy.rs new file mode 100644 index 0000000..08b717c --- /dev/null +++ b/src/team_policy.rs @@ -0,0 +1,333 @@ +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Map, Value}; + +pub const LOCK_MINUTES: [i32; 6] = [0, 5, 15, 30, 60, 240]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct LockPolicy { + pub max_minutes: i32, + pub force_vault: bool, +} + +impl LockPolicy { + pub fn from_columns(max_minutes: Option, force_vault: bool) -> Option { + max_minutes.map(|max_minutes| Self { + max_minutes, + force_vault, + }) + } +} + +pub const MACHINE_ONLY_KEYS: [&str; 5] = [ + "server.url", + "server.allowOther", + "server.previousUrls", + "enrollment.token", + "plugins.install", +]; + +pub const LOCK_POLICY_BACKFILL: &str = "UPDATE teams SET policy = jsonb_build_object('version', 1, 'policy', jsonb_build_object('security.lockTimeoutMinutes', jsonb_build_object('$range', jsonb_build_object('max', lock_max_minutes))) || CASE WHEN lock_force_vault THEN jsonb_build_object('security.lockAction', 'vault') ELSE '{}'::jsonb END), policy_revision = 1 WHERE lock_max_minutes IS NOT NULL AND policy IS NULL;"; + +const MAX_BYTES: usize = 65_536; +const MAX_KEYS: usize = 500; +const MAX_KEY_LEN: usize = 128; +const TIMEOUT_KEY: &str = "security.lockTimeoutMinutes"; +const ACTION_KEY: &str = "security.lockAction"; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DocError { + Invalid(String), + TooLarge, + DefaultsUnsupported, + MachineOnlyKey(String), +} + +impl DocError { + pub fn code(&self) -> &'static str { + match self { + Self::Invalid(_) => "POLICY_INVALID", + Self::TooLarge => "POLICY_TOO_LARGE", + Self::DefaultsUnsupported => "TEAM_DEFAULTS_UNSUPPORTED", + Self::MachineOnlyKey(_) => "POLICY_MACHINE_ONLY_KEY", + } + } + + pub fn key(&self) -> Option<&str> { + match self { + Self::MachineOnlyKey(k) => Some(k), + _ => None, + } + } +} + +fn invalid(msg: &str) -> DocError { + DocError::Invalid(msg.to_string()) +} + +fn valid_key(k: &str) -> bool { + (1..=MAX_KEY_LEN).contains(&k.len()) + && k.bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'.' | b'-')) +} + +fn valid_range(v: &Value) -> bool { + let Some(r) = v.as_object() else { return false }; + r.iter().all(|(k, v)| match k.as_str() { + "min" | "max" => v.is_number(), + "allowed" => v.is_array(), + _ => false, + }) +} + +pub fn validate_doc(doc: &Value) -> Result<(), DocError> { + let bytes = serde_json::to_vec(doc).map_err(|_| invalid("unserializable"))?; + if bytes.len() > MAX_BYTES { + return Err(DocError::TooLarge); + } + let obj = doc.as_object().ok_or_else(|| invalid("not an object"))?; + if obj.get("version") != Some(&json!(1)) { + return Err(invalid("version must be 1")); + } + match obj.get("defaults") { + None => {} + Some(Value::Object(d)) if d.is_empty() => {} + Some(Value::Object(_)) => return Err(DocError::DefaultsUnsupported), + Some(_) => return Err(invalid("defaults must be an object")), + } + let policy = obj + .get("policy") + .and_then(Value::as_object) + .ok_or_else(|| invalid("policy must be an object"))?; + if policy.len() > MAX_KEYS { + return Err(invalid("too many policy keys")); + } + for (k, v) in policy { + if MACHINE_ONLY_KEYS.contains(&k.as_str()) { + return Err(DocError::MachineOnlyKey(k.clone())); + } + if !valid_key(k) { + return Err(invalid("bad policy key")); + } + if let Some(range) = v.as_object().and_then(|o| o.get("$range")) { + if !valid_range(range) { + return Err(invalid("bad $range")); + } + } + } + Ok(()) +} + +pub fn with_lock_policy(doc: Option, lp: Option) -> Option { + let mut doc = doc.unwrap_or_else(|| json!({"version": 1, "policy": {}})); + let policy = doc.get_mut("policy").and_then(Value::as_object_mut)?; + policy.remove(TIMEOUT_KEY); + policy.remove(ACTION_KEY); + if let Some(lp) = lp { + policy.insert(TIMEOUT_KEY.into(), json!({"$range": {"max": lp.max_minutes}})); + if lp.force_vault { + policy.insert(ACTION_KEY.into(), json!("vault")); + } + } + (!policy.is_empty()).then_some(doc) +} + +pub fn lock_policy_of(doc: Option<&Value>) -> Option { + let policy = policy_map(doc)?; + let max = policy.get(TIMEOUT_KEY)?.get("$range")?.get("max")?.as_f64()?; + let max_minutes = LOCK_MINUTES + .iter() + .rev() + .copied() + .find(|m| f64::from(*m) <= max) + .unwrap_or(0); + Some(LockPolicy { + max_minutes, + force_vault: policy.get(ACTION_KEY) == Some(&json!("vault")), + }) +} + +fn policy_map(doc: Option<&Value>) -> Option<&Map> { + doc?.get("policy")?.as_object() +} + +pub fn is_narrowing(stored: Option<&Value>, new: &Value) -> bool { + let Some(new) = policy_map(Some(new)) else { + return false; + }; + let stored = policy_map(stored); + new.iter() + .all(|(k, v)| stored.and_then(|s| s.get(k)) == Some(v)) +} + +pub fn key_diff(old: Option<&Value>, new: Option<&Value>) -> Value { + let (old, new) = (policy_map(old), policy_map(new)); + let keys: BTreeSet<&String> = old.into_iter().chain(new).flat_map(|m| m.keys()).collect(); + let (mut added, mut changed, mut removed) = (vec![], vec![], vec![]); + for k in keys { + match (old.and_then(|m| m.get(k)), new.and_then(|m| m.get(k))) { + (None, Some(_)) => added.push(k), + (Some(_), None) => removed.push(k), + (Some(a), Some(b)) if a != b => changed.push(k), + _ => {} + } + } + json!({"added": added, "changed": changed, "removed": removed}) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_support::{seed_team_with_roles, seed_user}; + + fn doc(policy: Value) -> Value { + json!({"version": 1, "policy": policy}) + } + + fn lp(max_minutes: i32, force_vault: bool) -> LockPolicy { + LockPolicy { + max_minutes, + force_vault, + } + } + + #[test] + fn accepts_minimal_doc() { + assert_eq!(validate_doc(&doc(json!({"sync.types.key": false}))), Ok(())); + } + + #[test] + fn rejects_bad_shapes() { + let long = "a".repeat(129); + let bad = [ + json!({"version": 2, "policy": {}}), + json!({"version": "1", "policy": {}}), + json!({"version": 1, "policy": []}), + doc(json!({"bad key": 1})), + doc(json!({ long: 1 })), + doc(json!({"a.b": {"$range": {"max": 5, "step": 1}}})), + doc(json!({"a.b": {"$range": {"max": "5"}}})), + ]; + for d in bad { + assert!(matches!(validate_doc(&d), Err(DocError::Invalid(_))), "{d}"); + } + } + + #[test] + fn rejects_defaults() { + let with = |defaults: Value| json!({"version": 1, "policy": {}, "defaults": defaults}); + assert_eq!( + validate_doc(&with(json!({"a.b": 1}))), + Err(DocError::DefaultsUnsupported) + ); + assert_eq!(validate_doc(&with(json!({}))), Ok(())); + } + + #[test] + fn rejects_each_machine_only_key() { + for k in MACHINE_ONLY_KEYS { + let err = validate_doc(&doc(json!({ k: 1 }))).unwrap_err(); + assert_eq!(err, DocError::MachineOnlyKey(k.to_string())); + assert_eq!(err.code(), "POLICY_MACHINE_ONLY_KEY"); + assert_eq!(err.key(), Some(k)); + } + } + + #[test] + fn rejects_oversize() { + let many: Map = (0..501).map(|i| (format!("k{i}"), json!(1))).collect(); + assert!(matches!( + validate_doc(&doc(Value::Object(many))), + Err(DocError::Invalid(_)) + )); + let big = doc(json!({"a.b": "x".repeat(70 * 1024)})); + assert_eq!(validate_doc(&big), Err(DocError::TooLarge)); + } + + #[test] + fn lock_policy_round_trip() { + let expected = doc(json!({ + "security.lockTimeoutMinutes": {"$range": {"max": 15}}, + "security.lockAction": "vault" + })); + let made = with_lock_policy(None, Some(lp(15, true))).unwrap(); + assert_eq!(made, expected); + assert_eq!(lock_policy_of(Some(&made)), Some(lp(15, true))); + assert_eq!(with_lock_policy(Some(made), None), None); + + let mut other = expected.clone(); + other["policy"]["sync.types.key"] = json!(false); + assert_eq!( + with_lock_policy(Some(other), None), + Some(doc(json!({"sync.types.key": false}))) + ); + } + + #[test] + fn lock_policy_of_rounds_down() { + let range = |max: i64| doc(json!({"security.lockTimeoutMinutes": {"$range": {"max": max}}})); + assert_eq!(lock_policy_of(Some(&range(20))), Some(lp(15, false))); + assert_eq!(lock_policy_of(Some(&range(3))), Some(lp(0, false))); + assert_eq!(lock_policy_of(Some(&doc(json!({"a.b": 1})))), None); + assert_eq!(lock_policy_of(None), None); + let screen = doc(json!({ + "security.lockTimeoutMinutes": {"$range": {"max": 5}}, + "security.lockAction": "screen" + })); + assert_eq!(lock_policy_of(Some(&screen)), Some(lp(5, false))); + } + + #[test] + fn narrowing() { + let stored = doc(json!({"a.b": 1, "c.d": 2})); + assert!(is_narrowing(Some(&stored), &doc(json!({"a.b": 1})))); + assert!(!is_narrowing(Some(&stored), &doc(json!({"a.b": 9})))); + assert!(!is_narrowing(Some(&stored), &doc(json!({"a.b": 1, "x.y": 1})))); + assert!(is_narrowing(None, &doc(json!({})))); + assert!(!is_narrowing(None, &doc(json!({"a.b": 1})))); + } + + #[test] + fn key_diff_lists_keys() { + let old = doc(json!({"a.b": 1, "c.d": 2, "e.f": 3})); + let new = doc(json!({"a.b": 1, "c.d": 9, "g.h": 4})); + assert_eq!( + key_diff(Some(&old), Some(&new)), + json!({"added": ["g.h"], "changed": ["c.d"], "removed": ["e.f"]}) + ); + assert_eq!( + key_diff(None, Some(&new)), + json!({"added": ["a.b", "c.d", "g.h"], "changed": [], "removed": []}) + ); + } + + #[test] + fn migration_contains_backfill() { + assert!(include_str!("../migrations/067_team_policy.sql").contains(LOCK_POLICY_BACKFILL)); + } + + #[tokio::test] + async fn boot_backfill_matches_with_lock_policy() { + let pool = crate::test_pool_or_skip!(); + let owner = seed_user(&pool).await; + let team = seed_team_with_roles(&pool, owner).await; + for force_vault in [true, false] { + sqlx::query( + "UPDATE teams SET lock_max_minutes = 15, lock_force_vault = $2, policy = NULL WHERE id = $1", + ) + .bind(team) + .bind(force_vault) + .execute(&pool) + .await + .unwrap(); + crate::db::backfill_team_policies(&pool).await; + let stored: Option = sqlx::query_scalar("SELECT policy FROM teams WHERE id = $1") + .bind(team) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stored, with_lock_policy(None, Some(lp(15, force_vault)))); + } + } +} diff --git a/src/test_support.rs b/src/test_support.rs index 1fad58a..038b91d 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -84,7 +84,7 @@ pub async fn test_pool() -> Option { .expect("connect to TEST_DATABASE_URL"); // sqlx's migrator takes a Postgres advisory lock, so concurrent test // invocations applying migrations is safe. - sqlx::migrate!("./migrations") + crate::db::migrator() .run(&pool) .await .expect("run migrations on test database"); @@ -314,6 +314,41 @@ pub async fn inviter_with(pool: &PgPool, team: Uuid, extra: i64) -> Uuid { member_with_role(pool, team, member | extra).await } +pub fn team_members_told( + rx: &mut tokio::sync::broadcast::Receiver, + team: Uuid, +) -> std::collections::HashSet { + let want = format!("team_members:{team}"); + let mut told = std::collections::HashSet::new(); + while let Ok(ev) = rx.try_recv() { + if let crate::sync_notifier::SyncEvent::BlobPushed { user_id, device_id } = ev { + if device_id == want { + told.insert(user_id); + } + } + } + told +} + +pub async fn team_audit_count(pool: &PgPool, team: Uuid, action: &str) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM audit_logs WHERE team_id = $1 AND action = $2") + .bind(team) + .bind(action) + .fetch_one(pool) + .await + .expect("count team audit events") +} + +pub async fn listed_team(pool: &PgPool, team: Uuid, user: Uuid) -> crate::routes::teams::TeamWithRole { + let axum::Json(teams) = crate::routes::teams::list_teams( + axum::extract::State(pool.clone()), + axum::Extension(crate::auth::AuthUser(user)), + ) + .await + .expect("list teams"); + teams.into_iter().find(|t| t.id == team).expect("team listed") +} + pub async fn set_user_tier(pool: &PgPool, user: Uuid, tier: &str) { sqlx::query("UPDATE users SET subscription_tier = $1 WHERE id = $2") .bind(tier)