Merge #311

311: RFC: read-only access JWT tokens r=psarna a=psarna

I wanted to quickly evaluate how we could add coarse-grained access control to sqld with JWT and came up with this short patch, I'd be grateful for any comments/opinions.

The mechanism of authenticating JWT is extended a little bit. So far we only cared if the token is there and is valid. What JWT offers is an easily extensible mechanism for "claims", which is more or less just a JSON object which allows you to define properties that the token bearer holds. This patch introduces another JWT claim, with a working name "access". If this field is set to "read-only", `sqld` will refuse to execute any requests except ones categorized as reads.

The change turned out to be extremely small and took ~30 minutes tests included, since thanks to `@MarinPostma's` work we already categorize queries anyway.

The class of use cases I have in mind here are browser apps. We certainly don't want to push full access tokens to users' browsers, but it might be just fine to "leak" a read-only token that lets you access the database, but not modify it. To bring a concrete example, one of my demo apps I use for testing sqld (http://sorry.idont.date) would no longer need to run on Cloudflare Workers and could instead fetch data straight from the browser - far edge!

Co-authored-by: Piotr Sarna <[email protected]>

Author: bors[bot]

scripts/gen_jwt.py

@@ -40,6 +40,10 @@
 }
 token = jwt.encode(claims, privkey_pem, "EdDSA")
 
+claims["a"] = "ro"
+ro_token = jwt.encode(claims, privkey_pem, "EdDSA")
+
 open("jwt_key.pem", "wb").write(pubkey_pem)
 open("jwt_key.base64", "wb").write(pubkey_base64)
-print(token)
+print(f"Full access: {token}")
+print(f"Read-only:   {ro_token}")

sqld/proto/proxy.proto

@@ -130,9 +130,15 @@ message OrCond {
     repeated Cond conds = 1;
 }
 
+enum Authorized {
+    READONLY = 0;
+    FULL = 1;
+}
+
 message ProgramReq {
     string client_id = 1;
     Program pgm = 2;
+    optional Authorized authorized = 3;
 }
 
 service Proxy {

sqld/src/auth.rs

@@ -37,17 +37,28 @@ pub enum AuthError {
     Other,
 }
 
+#[non_exhaustive]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Authorized {
+    FullAccess,
+    ReadOnly,
+}
+
 /// A witness that the user has been authenticated.
-#[derive(Debug)]
-pub struct Authenticated(());
+#[non_exhaustive]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Authenticated {
+    Anonymous,
+    Authorized(Authorized),
+}
 
 impl Auth {
     pub fn authenticate_http(
         &self,
         auth_header: Option<&hyper::header::HeaderValue>,
     ) -> Result<Authenticated, AuthError> {
         if self.disabled {
-            return Ok(Authenticated(()));
+            return Ok(Authenticated::Authorized(Authorized::FullAccess));
         }
 
         let Some(auth_header) = auth_header else {
@@ -64,7 +75,7 @@ impl Auth {
                 let actual_value = actual_value.trim_end_matches('=');
                 let expected_value = expected_value.trim_end_matches('=');
                 if actual_value == expected_value {
-                    Ok(Authenticated(()))
+                    Ok(Authenticated::Authorized(Authorized::FullAccess))
                 } else {
                     Err(AuthError::BasicRejected)
                 }
@@ -75,7 +86,7 @@ impl Auth {
 
     pub fn authenticate_jwt(&self, jwt: Option<&str>) -> Result<Authenticated, AuthError> {
         if self.disabled {
-            return Ok(Authenticated(()));
+            return Ok(Authenticated::Authorized(Authorized::FullAccess));
         }
 
         let Some(jwt) = jwt else {
@@ -128,8 +139,18 @@ fn validate_jwt(
     let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::EdDSA);
     validation.required_spec_claims.remove("exp");
 
-    match jsonwebtoken::decode::<serde_json::Value>(jwt, jwt_key, &validation) {
-        Ok(_token) => Ok(Authenticated(())),
+    match jsonwebtoken::decode::<serde_json::Value>(jwt, jwt_key, &validation).map(|t| t.claims) {
+        Ok(serde_json::Value::Object(claims)) => {
+            tracing::trace!("Claims: {claims:#?}");
+            Ok(match claims.get("a").and_then(|s| s.as_str()) {
+                Some("ro") => Authenticated::Authorized(Authorized::ReadOnly),
+                Some("rw") => Authenticated::Authorized(Authorized::FullAccess),
+                Some(_) => Authenticated::Anonymous,
+                // Backward compatibility - no access claim means full access
+                None => Authenticated::Authorized(Authorized::FullAccess),
+            })
+        }
+        Ok(_) => Err(AuthError::JwtInvalid),
         Err(error) => Err(match error.kind() {
             ErrorKind::InvalidToken
             | ErrorKind::InvalidSignature
@@ -201,10 +222,13 @@ mod tests {
         auth.authenticate_http(Some(&HeaderValue::from_str(header).unwrap()))
     }
 
-    const VALID_JWT: &str = "eyJ0eXAiOiJKV1QiLCJhbGciOiJFZERTQSJ9.\
-        eyJleHAiOjQ4MzEwOTI5NDh9.\
-        TbPFJBxqb0fPPXj74DgmIZO41skmNEx-8b3PfAXv7IJMeLa3fNgBi7J5xxLm_-0SMEV3f6KMgUN0dBFbGRk4Ag";
-    const VALID_JWT_KEY: &str = "3dwzg2D96T4GcyZkK4MezpRQxU321g7aTrUn1iwOF0s";
+    const VALID_JWT_KEY: &str = "zaMv-aFGmB7PXkjM4IrMdF6B5zCYEiEGXW3RgMjNAtc";
+    const VALID_JWT: &str = "eyJhbGciOiJFZERTQSIsInR5cCI6IkpXVCJ9.\
+        eyJleHAiOjc5ODg0ODM4Mjd9.\
+        MatB2aLnPFusagqH2RMoVExP37o2GFLmaJbmd52OdLtAehRNeqeJZPrefP1t2GBFidApUTLlaBRL6poKq_s3CQ";
+    const VALID_READONLY_JWT: &str = "eyJhbGciOiJFZERTQSIsInR5cCI6IkpXVCJ9.\
+        eyJleHAiOjc5ODg0ODM4MjcsImEiOiJybyJ9.\
+        _2ZZiO2HC8b3CbCHSCufXXBmwpl-dLCv5O9Owvpy7LZ9aiQhXODpgV-iCdTsLQJ5FVanWhfn3FtJSnmWHn25DQ";
 
     macro_rules! assert_ok {
         ($e:expr) => {
@@ -268,6 +292,11 @@ mod tests {
             &auth,
             &format!("Bearer {}", &VALID_JWT[..80])
         ));
+
+        assert_eq!(
+            authenticate_http(&auth, &format!("Bearer {VALID_READONLY_JWT}")).unwrap(),
+            Authenticated::Authorized(Authorized::ReadOnly)
+        );
     }
 
     #[test]

sqld/src/database/libsql.rs

@@ -7,10 +7,11 @@ use rusqlite::{OpenFlags, StatementStatus};
 use tokio::sync::oneshot;
 use tracing::warn;
 
+use crate::auth::{Authenticated, Authorized};
 use crate::error::Error;
 use crate::libsql::wal_hook::WalHook;
 use crate::query::{Column, Query, QueryResponse, QueryResult, ResultSet, Row};
-use crate::query_analysis::{State, Statement};
+use crate::query_analysis::{State, Statement, StmtKind};
 use crate::stats::Stats;
 use crate::Result;
 
@@ -344,9 +345,37 @@ fn eval_cond(cond: &Cond, results: &[Option<QueryResult>]) -> Result<bool> {
     })
 }
 
+fn check_auth(auth: Authenticated, pgm: &Program) -> Result<()> {
+    for step in pgm.steps() {
+        let query = &step.query;
+        match (query.stmt.kind, &auth) {
+            (_, Authenticated::Anonymous) => {
+                return Err(Error::NotAuthorized(
+                    "anonymous access not allowed".to_string(),
+                ));
+            }
+            (StmtKind::Read, Authenticated::Authorized(_)) => (),
+            (StmtKind::TxnBegin, _) | (StmtKind::TxnEnd, _) => (),
+            (_, Authenticated::Authorized(Authorized::FullAccess)) => (),
+            _ => {
+                return Err(Error::NotAuthorized(format!(
+                    "Current session is not authorized to run: {}",
+                    query.stmt.stmt
+                )));
+            }
+        }
+    }
+    Ok(())
+}
+
 #[async_trait::async_trait]
 impl Database for LibSqlDb {
-    async fn execute_program(&self, pgm: Program) -> Result<(Vec<Option<QueryResult>>, State)> {
+    async fn execute_program(
+        &self,
+        pgm: Program,
+        auth: Authenticated,
+    ) -> Result<(Vec<Option<QueryResult>>, State)> {
+        check_auth(auth, &pgm)?;
         let (resp, receiver) = oneshot::channel();
         let msg = Message { pgm, resp };
         let _ = self.sender.send(msg);

sqld/src/database/mod.rs

@@ -1,5 +1,6 @@
 use std::sync::Arc;
 
+use crate::auth::Authenticated;
 use crate::query::{Params, Query, QueryResult};
 use crate::query_analysis::{State, Statement};
 use crate::Result;
@@ -50,13 +51,17 @@ pub enum Cond {
 #[async_trait::async_trait]
 pub trait Database: Send + Sync {
     /// Executes a query program
-    async fn execute_program(&self, pgm: Program) -> Result<(Vec<Option<QueryResult>>, State)>;
+    async fn execute_program(
+        &self,
+        pgm: Program,
+        auth: Authenticated,
+    ) -> Result<(Vec<Option<QueryResult>>, State)>;
 
     /// Unconditionnaly execute a query as part of a program
-    async fn execute_one(&self, query: Query) -> Result<(QueryResult, State)> {
+    async fn execute_one(&self, query: Query, auth: Authenticated) -> Result<(QueryResult, State)> {
         let pgm = Program::new(vec![Step { cond: None, query }]);
 
-        let (results, state) = self.execute_program(pgm).await?;
+        let (results, state) = self.execute_program(pgm, auth).await?;
         Ok((results.into_iter().next().unwrap().unwrap(), state))
     }
 
@@ -66,6 +71,7 @@ pub trait Database: Send + Sync {
     async fn execute_batch_or_rollback(
         &self,
         batch: Vec<Query>,
+        auth: Authenticated,
     ) -> Result<(Vec<Option<QueryResult>>, State)> {
         let mut steps = make_batch_program(batch);
 
@@ -86,7 +92,7 @@ pub trait Database: Send + Sync {
 
         let pgm = Program::new(steps);
 
-        let (mut results, state) = self.execute_program(pgm).await?;
+        let (mut results, state) = self.execute_program(pgm, auth).await?;
         // remove the rollback result
         results.pop();
 
@@ -95,18 +101,25 @@ pub trait Database: Send + Sync {
 
     /// Execute all the queries in the batch sequentially.
     /// If an query in the batch fails, the remaining queries are ignored
-    async fn execute_batch(&self, batch: Vec<Query>) -> Result<(Vec<Option<QueryResult>>, State)> {
+    async fn execute_batch(
+        &self,
+        batch: Vec<Query>,
+        auth: Authenticated,
+    ) -> Result<(Vec<Option<QueryResult>>, State)> {
         let steps = make_batch_program(batch);
         let pgm = Program::new(steps);
-        self.execute_program(pgm).await
+        self.execute_program(pgm, auth).await
     }
 
-    async fn rollback(&self) -> Result<()> {
+    async fn rollback(&self, auth: Authenticated) -> Result<()> {
         let (results, _) = self
-            .execute_one(Query {
-                stmt: Statement::parse("ROLLBACK").next().unwrap().unwrap(),
-                params: Params::empty(),
-            })
+            .execute_one(
+                Query {
+                    stmt: Statement::parse("ROLLBACK").next().unwrap().unwrap(),
+                    params: Params::empty(),
+                },
+                auth,
+            )
             .await?;
 
         results?;

sqld/src/database/write_proxy.rs

@@ -5,6 +5,7 @@ use tokio::sync::Mutex;
 use tonic::transport::Channel;
 use uuid::Uuid;
 
+use crate::auth::{Authenticated, Authorized};
 use crate::error::Error;
 use crate::query::{QueryResponse, QueryResult};
 use crate::query_analysis::State;
@@ -74,11 +75,18 @@ impl WriteProxyDatabase {
         &self,
         pgm: Program,
         state: &mut State,
+        auth: Authenticated,
     ) -> Result<(Vec<Option<QueryResult>>, State)> {
         let mut client = self.write_proxy.clone();
+        let authorized: Option<i32> = match auth {
+            Authenticated::Anonymous => None,
+            Authenticated::Authorized(Authorized::ReadOnly) => Some(0),
+            Authenticated::Authorized(Authorized::FullAccess) => Some(1),
+        };
         let req = crate::rpc::proxy::rpc::ProgramReq {
             client_id: self.client_id.to_string(),
             pgm: Some(pgm.into()),
+            authorized,
         };
         match client.execute(req).await {
             Ok(r) => {
@@ -110,21 +118,25 @@ impl WriteProxyDatabase {
 
 #[async_trait::async_trait]
 impl Database for WriteProxyDatabase {
-    async fn execute_program(&self, pgm: Program) -> Result<(Vec<Option<QueryResult>>, State)> {
+    async fn execute_program(
+        &self,
+        pgm: Program,
+        auth: Authenticated,
+    ) -> Result<(Vec<Option<QueryResult>>, State)> {
         let mut state = self.state.lock().await;
         if *state == State::Init && pgm.is_read_only() {
             // We know that this program won't perform any writes. We attempt to run it on the
             // replica. If it leaves an open transaction, then this program is an interactive
             // transaction, so we rollback the replica, and execute again on the primary.
-            let (results, new_state) = self.read_db.execute_program(pgm.clone()).await?;
+            let (results, new_state) = self.read_db.execute_program(pgm.clone(), auth).await?;
             if new_state != State::Init {
-                self.read_db.rollback().await?;
-                self.execute_remote(pgm, &mut state).await
+                self.read_db.rollback(auth).await?;
+                self.execute_remote(pgm, &mut state, auth).await
             } else {
                 Ok((results, new_state))
             }
         } else {
-            self.execute_remote(pgm, &mut state).await
+            self.execute_remote(pgm, &mut state, auth).await
         }
     }
 }

sqld/src/error.rs

@@ -23,6 +23,8 @@ pub enum Error {
     Internal(String),
     #[error("Invalid batch step: {0}")]
     InvalidBatchStep(usize),
+    #[error("Not authorized to execute query: {0}")]
+    NotAuthorized(String),
 }
 
 impl From<tokio::sync::oneshot::error::RecvError> for Error {

sqld/src/hrana/batch.rs

@@ -1,5 +1,6 @@
 use anyhow::Result;
 
+use crate::auth::Authenticated;
 use crate::database::{Cond, Database, Program, Step};
 
 use super::proto;
@@ -71,11 +72,15 @@ fn batch_to_program(batch: &proto::Batch) -> Result<Program> {
     Ok(Program::new(steps))
 }
 
-pub async fn execute_batch(db: &dyn Database, batch: &proto::Batch) -> Result<proto::BatchResult> {
+pub async fn execute_batch(
+    db: &dyn Database,
+    auth: Authenticated,
+    batch: &proto::Batch,
+) -> Result<proto::BatchResult> {
     let pgm = batch_to_program(batch)?;
     let mut step_results = Vec::with_capacity(pgm.steps.len());
     let mut step_errors = Vec::with_capacity(pgm.steps.len());
-    let (results, _state) = db.execute_program(pgm).await?;
+    let (results, _state) = db.execute_program(pgm, auth).await?;
     for result in results {
         let (step_result, step_error) = match result {
             Some(Ok(r)) => (Some(proto_stmt_result_from_query_response(r)), None),