rws_server/authz/
sql.rs

1//! [`SqlStore`]: relationship tuples in a database (SQLite, Postgres or
2//! MySQL through the model layer), with a short read cache.
3
4use std::collections::HashMap;
5use std::sync::{Arc, Mutex, OnceLock};
6use std::time::{Duration, Instant};
7
8use super::{AuthzError, ObjectRef, Subject, Tuple, TupleFilter, TupleStore};
9use crate::model::backend::Backend;
10use crate::model::{DbConfig, DbPool, Value};
11
12/// How long a read is reused (this process's own writes clear it at once).
13pub const READ_CACHE: Duration = Duration::from_secs(1);
14
15const TABLE: &str = "rws_authz_tuples";
16
17/// The runtime the store's queries run on: the pool's connections belong to
18/// it, whatever runtime (or none) the caller is on.
19fn runtime() -> &'static tokio::runtime::Runtime {
20    static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
21    RUNTIME.get_or_init(|| {
22        tokio::runtime::Builder::new_multi_thread()
23            .worker_threads(2)
24            .thread_name("rws-authz-sql")
25            .enable_all()
26            .build()
27            .expect("the authorization store runtime")
28    })
29}
30
31/// Run `f` on the store's runtime and wait for it (from any thread that
32/// isn't one of its workers).
33#[doc(hidden)]
34pub fn run<T: Send + 'static>(f: impl std::future::Future<Output = T> + Send + 'static) -> Result<T, AuthzError> {
35    let (tx, rx) = std::sync::mpsc::sync_channel(1);
36    runtime().spawn(async move {
37        let _ = tx.send(f.await);
38    });
39    rx.recv().map_err(|_| AuthzError::Store("the query was dropped".to_string()))
40}
41
42fn store_error(e: impl std::fmt::Display) -> AuthzError {
43    AuthzError::Store(e.to_string())
44}
45
46/// Tuples in the `rws_authz_tuples` table (created if missing): one row per
47/// tuple, keyed by the SHA-256 of its text so the key fits every database's
48/// index limit, indexed by object and relation for checks.
49pub struct SqlStore {
50    pool: DbPool,
51    cache: Mutex<HashMap<(ObjectRef, String), (Instant, Vec<Subject>)>>,
52}
53
54impl SqlStore {
55    /// Connect with `config` (its pool lives on the store's own runtime) and
56    /// create the table if it's missing.
57    pub fn open(config: DbConfig) -> Result<SqlStore, AuthzError> {
58        let pool = run(async move { DbPool::new(config).await })?.map_err(store_error)?;
59        SqlStore::new(pool)
60    }
61
62    /// [`open`](Self::open) with `RWS_DB_*` from the environment, TLS
63    /// (`RWS_DB_SSL_*`) included.
64    pub fn from_env() -> Result<SqlStore, AuthzError> {
65        let config = DbConfig::from_env().map_err(store_error)?;
66        let tls = crate::model::DbTls::from_env().map_err(store_error)?;
67        let pool = run(async move { DbPool::new_with_tls(config, tls).await })?.map_err(store_error)?;
68        SqlStore::new(pool)
69    }
70
71    /// Use `pool` and create the table if it's missing. Queries run on the
72    /// store's runtime: a pool made on another runtime must outlive it.
73    pub fn new(pool: DbPool) -> Result<SqlStore, AuthzError> {
74        let store = SqlStore { pool, cache: Mutex::default() };
75        let text = if matches!(store.pool.backend(), b if is_mysql(b)) { "VARCHAR(256) CHARACTER SET utf8mb4" } else { "VARCHAR(256)" };
76        let create = format!(
77            "CREATE TABLE IF NOT EXISTS {TABLE} (tuple_key CHAR(64) NOT NULL PRIMARY KEY, object_type {text} NOT NULL, \
78             object_id {text} NOT NULL, relation {text} NOT NULL, subject_type {text} NOT NULL, subject_id {text} NOT NULL, \
79             subject_relation {text} NOT NULL)"
80        );
81        store.execute(create, Vec::new())?;
82        let index = if is_mysql(store.pool.backend()) {
83            // MySQL has no IF NOT EXISTS for indexes: a duplicate is fine.
84            format!("CREATE INDEX {TABLE}_object ON {TABLE} (object_type(64), object_id(128), relation(64))")
85        } else {
86            format!("CREATE INDEX IF NOT EXISTS {TABLE}_object ON {TABLE} (object_type, object_id, relation)")
87        };
88        if let Err(e) = store.execute(index, Vec::new()) {
89            if !is_mysql(store.pool.backend()) {
90                return Err(e);
91            }
92        }
93        Ok(store)
94    }
95
96    fn execute(&self, sql: String, params: Vec<Value>) -> Result<u64, AuthzError> {
97        let pool = self.pool.clone();
98        run(async move { pool.execute(&sql, &params).await })?.map_err(store_error)
99    }
100
101    fn query(&self, sql: String, params: Vec<Value>) -> Result<Vec<crate::model::ModelRow>, AuthzError> {
102        let pool = self.pool.clone();
103        run(async move { pool.query_rows(&sql, &params).await })?.map_err(store_error)
104    }
105
106    /// The `n`th placeholder (from 1) in this database's syntax.
107    fn placeholder(&self, n: usize) -> String {
108        if is_postgres(self.pool.backend()) { format!("${n}") } else { "?".to_string() }
109    }
110
111    fn forget(&self) {
112        self.cache.lock().unwrap_or_else(|e| e.into_inner()).clear();
113    }
114}
115
116fn is_mysql(backend: Backend) -> bool {
117    #[cfg(feature = "model-mysql")]
118    if matches!(backend, Backend::MySql) {
119        return true;
120    }
121    let _ = backend;
122    false
123}
124
125fn is_postgres(backend: Backend) -> bool {
126    #[cfg(feature = "model-postgres")]
127    if matches!(backend, Backend::Postgres) {
128        return true;
129    }
130    let _ = backend;
131    false
132}
133
134fn key(tuple: &Tuple) -> String {
135    crate::core::sha256::sha256(tuple.to_string().as_bytes()).iter().map(|b| format!("{b:02x}")).collect()
136}
137
138fn subject_from(row: &crate::model::ModelRow) -> Result<Subject, AuthzError> {
139    let kind: String = row.get("subject_type").map_err(store_error)?;
140    let id: String = row.get("subject_id").map_err(store_error)?;
141    let relation: String = row.get("subject_relation").map_err(store_error)?;
142    Ok(Subject { object: ObjectRef { kind, id }, relation: (!relation.is_empty()).then_some(relation) })
143}
144
145impl TupleStore for SqlStore {
146    fn write(&self, tuple: &Tuple) -> Result<bool, AuthzError> {
147        let backend = self.pool.backend();
148        let values: Vec<String> = (1..=7).map(|n| self.placeholder(n)).collect();
149        let columns = "tuple_key, object_type, object_id, relation, subject_type, subject_id, subject_relation";
150        let sql = if is_mysql(backend) {
151            format!("INSERT IGNORE INTO {TABLE} ({columns}) VALUES ({})", values.join(", "))
152        } else if is_postgres(backend) {
153            format!("INSERT INTO {TABLE} ({columns}) VALUES ({}) ON CONFLICT DO NOTHING", values.join(", "))
154        } else {
155            format!("INSERT OR IGNORE INTO {TABLE} ({columns}) VALUES ({})", values.join(", "))
156        };
157        let params = vec![
158            Value::Text(key(tuple)),
159            Value::Text(tuple.object.kind.clone()),
160            Value::Text(tuple.object.id.clone()),
161            Value::Text(tuple.relation.clone()),
162            Value::Text(tuple.subject.object.kind.clone()),
163            Value::Text(tuple.subject.object.id.clone()),
164            Value::Text(tuple.subject.relation.clone().unwrap_or_default()),
165        ];
166        let inserted = self.execute(sql, params)? > 0;
167        self.forget();
168        Ok(inserted)
169    }
170
171    fn delete(&self, tuple: &Tuple) -> Result<bool, AuthzError> {
172        let sql = format!("DELETE FROM {TABLE} WHERE tuple_key = {}", self.placeholder(1));
173        let removed = self.execute(sql, vec![Value::Text(key(tuple))])? > 0;
174        self.forget();
175        Ok(removed)
176    }
177
178    fn read(&self, object: &ObjectRef, relation: &str) -> Result<Vec<Subject>, AuthzError> {
179        let cache_key = (object.clone(), relation.to_string());
180        if let Some((at, subjects)) = self.cache.lock().unwrap_or_else(|e| e.into_inner()).get(&cache_key) {
181            if at.elapsed() < READ_CACHE {
182                return Ok(subjects.clone());
183            }
184        }
185        let sql = format!(
186            "SELECT subject_type, subject_id, subject_relation FROM {TABLE} WHERE object_type = {} AND object_id = {} AND relation = {}",
187            self.placeholder(1),
188            self.placeholder(2),
189            self.placeholder(3)
190        );
191        let rows = self.query(sql, vec![Value::Text(object.kind.clone()), Value::Text(object.id.clone()), Value::Text(relation.to_string())])?;
192        let subjects = rows.iter().map(subject_from).collect::<Result<Vec<_>, _>>()?;
193        let mut cache = self.cache.lock().unwrap_or_else(|e| e.into_inner());
194        if cache.len() > 10_000 {
195            cache.clear();
196        }
197        cache.insert(cache_key, (Instant::now(), subjects.clone()));
198        Ok(subjects)
199    }
200
201    fn list(&self, filter: &TupleFilter, limit: usize) -> Result<Vec<Tuple>, AuthzError> {
202        let mut conditions = Vec::new();
203        let mut params = Vec::new();
204        let mut add = |column: &str, value: String, conditions: &mut Vec<String>| {
205            params.push(Value::Text(value));
206            conditions.push(format!("{column} = {}", self.placeholder(params.len())));
207        };
208        if let Some(object) = &filter.object {
209            add("object_type", object.kind.clone(), &mut conditions);
210            add("object_id", object.id.clone(), &mut conditions);
211        }
212        if let Some(kind) = &filter.object_kind {
213            add("object_type", kind.clone(), &mut conditions);
214        }
215        if let Some(relation) = &filter.relation {
216            add("relation", relation.clone(), &mut conditions);
217        }
218        if let Some(subject) = &filter.subject {
219            add("subject_type", subject.object.kind.clone(), &mut conditions);
220            add("subject_id", subject.object.id.clone(), &mut conditions);
221            add("subject_relation", subject.relation.clone().unwrap_or_default(), &mut conditions);
222        }
223        let filter_sql = if conditions.is_empty() { String::new() } else { format!(" WHERE {}", conditions.join(" AND ")) };
224        let sql = format!(
225            "SELECT object_type, object_id, relation, subject_type, subject_id, subject_relation FROM {TABLE}{filter_sql} \
226             ORDER BY object_type, object_id, relation, subject_type, subject_id, subject_relation LIMIT {}",
227            limit.min(100_000)
228        );
229        let rows = self.query(sql, params)?;
230        rows.iter()
231            .map(|row| {
232                Ok(Tuple {
233                    object: ObjectRef { kind: row.get("object_type").map_err(store_error)?, id: row.get("object_id").map_err(store_error)? },
234                    relation: row.get("relation").map_err(store_error)?,
235                    subject: subject_from(row)?,
236                })
237            })
238            .collect()
239    }
240}
241
242/// The config file's SQL store (`store = "sql"`, `RWS_DB_*`), opened once
243/// per process.
244#[doc(hidden)]
245pub fn config_store() -> Result<Arc<dyn TupleStore>, String> {
246    static STORE: OnceLock<Result<Arc<SqlStore>, String>> = OnceLock::new();
247    STORE
248        .get_or_init(|| SqlStore::from_env().map(Arc::new).map_err(|e| e.to_string()))
249        .clone()
250        .map(|store| store as Arc<dyn TupleStore>)
251}
252
253#[cfg(test)]
254mod tests;
rws_server/authz/sql.rs

rws_server/authz/
sql.rs
