feat(auth): jwt validation and auth-aware service wrapper

Author: matt-codecov

Cargo.lock

@@ -19,6 +19,7 @@ console = "0.16.1"
 elegant-departure = { version = "0.3.2", features = ["tokio"] }
 figment = { version = "0.10.19", features = ["env", "test", "yaml"] }
 futures-util = { workspace = true }
+jsonwebtoken = { version = "10.2.0", features = ["rust_crypto"] }
 merni = { workspace = true }
 mimalloc = { workspace = true }
 num_cpus = "1.17.0"
@@ -36,6 +37,7 @@ secrecy = { version = "0.10.3", features = ["serde"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = "3.14.1"
+thiserror = "2.0.17"
 tokio = { workspace = true, features = ["full"] }
 tokio-stream = { workspace = true }
 tower = { version = "0.5.2" }

objectstore-server/Cargo.toml

@@ -0,0 +1,84 @@
+use axum::extract::FromRequestParts;
+use axum::http::{StatusCode, header, request::Parts};
+use objectstore_service::BackendStream;
+use objectstore_service::{ObjectPath, StorageService};
+use objectstore_types::Metadata;
+
+use super::{AuthContext, AuthError, Permission};
+use crate::state::ServiceState;
+
+pub struct AuthAwareService {
+    service: StorageService,
+
+    enforce: bool,
+
+    auth_context: Option<AuthContext>,
+}
+
+impl AuthAwareService {
+    fn assert_authorized(&self, perm: Permission, path: &ObjectPath) -> anyhow::Result<()> {
+        let auth_result = self
+            .auth_context
+            .as_ref()
+            .ok_or(AuthError::InvalidToken)
+            .and_then(|ac| ac.assert_authorized(perm, path));
+        if self.enforce {
+            return Ok(auth_result?);
+        }
+        Ok(())
+    }
+
+    /// Auth-aware wrapper around [`StorageService::put_object`].
+    pub async fn put_object(
+        &self,
+        path: ObjectPath,
+        metadata: &Metadata,
+        stream: BackendStream,
+    ) -> anyhow::Result<ObjectPath> {
+        self.assert_authorized(Permission::ObjectWrite, &path)?;
+
+        self.service.put_object(path, metadata, stream).await
+    }
+
+    /// Auth-aware wrapper around [`StorageService::get_object`].
+    pub async fn get_object(
+        &self,
+        path: &ObjectPath,
+    ) -> anyhow::Result<Option<(Metadata, BackendStream)>> {
+        self.assert_authorized(Permission::ObjectRead, path)?;
+
+        self.service.get_object(path).await
+    }
+
+    /// Auth-aware wrapper around [`StorageService::delete_object`].
+    pub async fn delete_object(&self, path: &ObjectPath) -> anyhow::Result<()> {
+        self.assert_authorized(Permission::ObjectDelete, path)?;
+
+        self.service.delete_object(path).await
+    }
+}
+
+impl FromRequestParts<ServiceState> for AuthAwareService {
+    type Rejection = StatusCode;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &ServiceState,
+    ) -> Result<Self, Self::Rejection> {
+        let encoded_token = parts
+            .headers
+            .get(header::AUTHORIZATION)
+            .and_then(|v| v.to_str().ok());
+
+        let auth_context = AuthContext::from_encoded_jwt(encoded_token, &state.config.auth);
+        if auth_context.is_err() && state.config.auth.enforce {
+            return Err(StatusCode::UNAUTHORIZED);
+        }
+
+        Ok(AuthAwareService {
+            service: state.service.clone(),
+            enforce: state.config.auth.enforce,
+            auth_context: auth_context.ok(),
+        })
+    }
+}

objectstore-server/src/auth.rs

@@ -0,0 +1,170 @@
+use std::collections::HashSet;
+
+use jsonwebtoken::{DecodingKey, Header, TokenData, Validation, decode, decode_header};
+use objectstore_service::ObjectPath;
+use secrecy::ExposeSecret;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+use crate::config::AuthZ;
+
+/// Permissions that control whether different operations are authorized.
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, Hash)]
+pub enum Permission {
+    /// The permission required to read objects from objectstore.
+    #[serde(rename = "object.read")]
+    ObjectRead,
+
+    /// The permission required to write/overwrite objects in objectstore.
+    #[serde(rename = "object.write")]
+    ObjectWrite,
+
+    /// The permission required to delete objects from objectstore.
+    #[serde(rename = "object.delete")]
+    ObjectDelete,
+}
+
+#[derive(Debug)]
+pub struct AuthContext {
+    pub usecase: String,
+
+    pub scope: Vec<String>,
+
+    pub permissions: HashSet<Permission>,
+
+    pub enforce: bool,
+
+    _phantom: std::marker::PhantomData<()>,
+}
+
+#[derive(Error, Debug)]
+pub enum AuthError {
+    #[error("could not process JWT")]
+    InvalidToken,
+
+    #[error("could not verify JWT signature")]
+    InvalidSignature,
+
+    #[error("operation not allowed")]
+    NotPermitted,
+}
+
+#[derive(Deserialize)]
+struct JwtClaims {
+    resource: Vec<String>,
+    permissions: HashSet<Permission>,
+}
+
+fn jwt_validation_params(jwt_header: &Header) -> Validation {
+    let mut validation = Validation::new(jwt_header.alg);
+    validation.set_audience(&["objectstore"]);
+    validation.set_issuer(&["sentry", "relay"]);
+    validation.set_required_spec_claims(&["exp"]);
+    validation
+}
+
+impl AuthContext {
+    pub fn from_encoded_jwt(
+        encoded_token: Option<&str>,
+        config: &AuthZ,
+    ) -> Result<AuthContext, AuthError> {
+        let Some(encoded_token) = encoded_token else {
+            tracing::debug!("No authorization token provided");
+            return Err(AuthError::InvalidToken);
+        };
+
+        let Ok(jwt_header) = decode_header(encoded_token) else {
+            tracing::debug!("Provided authorization token is not valid JWT");
+            return Err(AuthError::InvalidToken);
+        };
+
+        let Some(key_id) = jwt_header.kid.as_ref() else {
+            tracing::debug!("JWT header is missing `kid` field");
+            return Err(AuthError::InvalidToken);
+        };
+
+        let Some(key_config) = config.keys.get(key_id) else {
+            return Err(AuthError::InvalidToken);
+        };
+
+        let mut verified_claims: Option<TokenData<JwtClaims>> = None;
+        for key in &key_config.key_versions {
+            // TODO: Certain failure reasons (e.g. token expiration) should early-exit
+            verified_claims = decode::<JwtClaims>(
+                encoded_token,
+                &DecodingKey::from_secret(key.expose_secret().as_bytes()),
+                &jwt_validation_params(&jwt_header),
+            )
+            .ok();
+            if verified_claims.is_some() {
+                break;
+            }
+        }
+
+        let Some(mut verified_claims) = verified_claims else {
+            tracing::debug!("Failed to verify JWT with all configured keys");
+            return Err(AuthError::InvalidSignature);
+        };
+
+        // The JWT's `resource` is `{usecase}/{*scope}`
+        let usecase = verified_claims.claims.resource.remove(0);
+        let scope = verified_claims.claims.resource;
+
+        // Taking the intersection here ensures the `AuthContext` does not have any permissions
+        // that `key_config.max_permissions` doesn't have, even if the token tried to grant them.
+        let permissions = verified_claims
+            .claims
+            .permissions
+            .intersection(&key_config.max_permissions)
+            .cloned()
+            .collect();
+
+        Ok(AuthContext {
+            usecase,
+            scope,
+            permissions,
+            enforce: config.enforce,
+            _phantom: std::marker::PhantomData,
+        })
+    }
+
+    fn fail_if_enforced(&self, perm: &Permission, path: &ObjectPath) -> Result<(), AuthError> {
+        tracing::debug!(?self, ?perm, ?path, "Authorization failed");
+        if self.enforce {
+            return Err(AuthError::NotPermitted);
+        }
+        Ok(())
+    }
+
+    pub fn assert_authorized(&self, perm: Permission, path: &ObjectPath) -> Result<(), AuthError> {
+        if !self.permissions.contains(&perm) {
+            self.fail_if_enforced(&perm, path)?;
+        }
+
+        // Ensures that the `AuthContext`'s scope is a prefix or exact match of the `ObjectPath`'s
+        // scope. For example:
+        //
+        // Allowed:
+        //     path.scope: state.wa / city.seattle
+        //     self.scope: state.wa / city.seattle
+        //
+        // Allowed:
+        //     path.scope: state.wa / city.seattle
+        //     self.scope: state.wa
+        //
+        // Not allowed:
+        //     path.scope: state.wa / city.seattle
+        //     self.scope: state.wa / city.seattle / neighborhood.fremont
+        //
+        // Not allowed:
+        //     path.scope: state.wa / city.seattle
+        //     self.scope: state.mi
+        for (auth_segment, obj_segment) in self.scope.iter().zip(path.scope.iter()) {
+            if auth_segment != obj_segment {
+                self.fail_if_enforced(&perm, path)?;
+            }
+        }
+
+        Ok(())
+    }
+}

objectstore-server/src/auth/auth_aware_service.rs

@@ -0,0 +1,5 @@
+mod auth_context;
+pub use auth_context::*;
+
+mod auth_aware_service;
+pub use auth_aware_service::*;

objectstore-server/src/auth/auth_context.rs

@@ -5,7 +5,7 @@ use std::time::SystemTime;
 
 use anyhow::Context;
 use axum::body::Body;
-use axum::extract::{Path, State};
+use axum::extract::Path;
 use axum::http::{HeaderMap, Method, StatusCode};
 use axum::response::{IntoResponse, Response};
 use axum::routing;
@@ -15,6 +15,7 @@ use objectstore_service::{ObjectPath, OptionalObjectPath};
 use objectstore_types::Metadata;
 use serde::Serialize;
 
+use crate::auth::AuthAwareService;
 use crate::error::ApiResult;
 use crate::state::ServiceState;
 
@@ -42,8 +43,8 @@ struct PutBlobResponse {
 }
 
 async fn insert_object(
-    State(state): State<ServiceState>,
     Path(path): Path<OptionalObjectPath>,
+    service: AuthAwareService,
     method: Method,
     headers: HeaderMap,
     body: Body,
@@ -66,7 +67,7 @@ async fn insert_object(
     metadata.time_created = Some(SystemTime::now());
 
     let stream = body.into_data_stream().map_err(io::Error::other).boxed();
-    let response_path = state.service.put_object(path, &metadata, stream).await?;
+    let response_path = service.put_object(path, &metadata, stream).await?;
     let response = Json(PutBlobResponse {
         key: response_path.key.to_string(),
     });
@@ -75,11 +76,11 @@ async fn insert_object(
 }
 
 async fn get_object(
-    State(state): State<ServiceState>,
+    service: AuthAwareService,
     Path(path): Path<ObjectPath>,
 ) -> ApiResult<Response> {
     populate_sentry_scope(&path);
-    let Some((metadata, stream)) = state.service.get_object(&path).await? else {
+    let Some((metadata, stream)) = service.get_object(&path).await? else {
         return Ok(StatusCode::NOT_FOUND.into_response());
     };
 
@@ -90,12 +91,12 @@ async fn get_object(
 }
 
 async fn delete_object(
-    State(state): State<ServiceState>,
+    service: AuthAwareService,
     Path(path): Path<ObjectPath>,
 ) -> ApiResult<impl IntoResponse> {
     populate_sentry_scope(&path);
 
-    state.service.delete_object(&path).await?;
+    service.delete_object(&path).await?;
 
     Ok(StatusCode::NO_CONTENT)
 }

objectstore-server/src/auth/mod.rs

@@ -14,6 +14,8 @@ use crate::ObjectPath;
 pub const USER_AGENT: &str = concat!("sentry-objectstore/", env!("CARGO_PKG_VERSION"));
 
 pub type BoxedBackend = Box<dyn Backend>;
+
+/// Type alias for data streams used in [`StorageService`] and [`Backend`] APIs.
 pub type BackendStream = BoxStream<'static, io::Result<Bytes>>;
 
 #[async_trait::async_trait]

objectstore-server/src/endpoints.rs

@@ -17,8 +17,9 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Instant;
 
-use crate::backend::common::{BackendStream, BoxedBackend};
+use crate::backend::common::BoxedBackend;
 
+pub use crate::backend::common::BackendStream;
 pub use path::*;
 
 /// The threshold up until which we will go to the "high volume" backend.