objectstore_server/extractors/
id.rs

1use std::borrow::Cow;
2
3use axum::extract::rejection::PathRejection;
4use axum::extract::{FromRequestParts, Path};
5use axum::http::request::Parts;
6use axum::response::{IntoResponse, Response};
7use objectstore_service::id::{ObjectContext, ObjectId};
8use objectstore_types::scope::{EMPTY_SCOPES, Scope, Scopes};
9use objectstore_types::usecase::Usecase;
10use serde::{Deserialize, de};
11
12use crate::extractors::Xt;
13use crate::extractors::downstream_service::DownstreamService;
14use crate::rate_limits::LimitExceeded;
15use crate::state::ServiceState;
16
17#[derive(Debug)]
18pub enum ObjectRejection {
19    Path(PathRejection),
20    Killswitched,
21    RateLimited(LimitExceeded),
22}
23
24impl IntoResponse for ObjectRejection {
25    fn into_response(self) -> Response {
26        match self {
27            ObjectRejection::Path(rejection) => rejection.into_response(),
28            ObjectRejection::Killswitched => (
29                http::StatusCode::FORBIDDEN,
30                "Object access is disabled for this scope through killswitches",
31            )
32                .into_response(),
33            ObjectRejection::RateLimited(error) => error.into_response(),
34        }
35    }
36}
37
38impl From<PathRejection> for ObjectRejection {
39    fn from(rejection: PathRejection) -> Self {
40        ObjectRejection::Path(rejection)
41    }
42}
43
44impl FromRequestParts<ServiceState> for Xt<ObjectId> {
45    type Rejection = ObjectRejection;
46
47    async fn from_request_parts(
48        parts: &mut Parts,
49        state: &ServiceState,
50    ) -> Result<Self, Self::Rejection> {
51        let Path(params) = Path::<ObjectParams>::from_request_parts(parts, state).await?;
52        let id = ObjectId::from_parts(params.usecase, params.scopes, params.key);
53
54        populate_sentry_context(id.context());
55        sentry::configure_scope(|s| s.set_extra("key", id.key().into()));
56
57        let service = DownstreamService::from_request_parts(parts, state)
58            .await
59            .unwrap();
60
61        if state
62            .config
63            .killswitches
64            .matches(id.context(), service.as_str())
65        {
66            return Err(ObjectRejection::Killswitched);
67        }
68
69        state
70            .rate_limiter
71            .check(id.context(), Some(id.key()))
72            .map_err(ObjectRejection::RateLimited)?;
73
74        Ok(Xt(id))
75    }
76}
77
78/// Path parameters used for object-level endpoints.
79///
80/// This is meant to be used with the axum `Path` extractor.
81#[derive(Clone, Debug, Deserialize)]
82struct ObjectParams {
83    usecase: Usecase,
84    #[serde(deserialize_with = "deserialize_scopes")]
85    scopes: Scopes,
86    key: String,
87}
88
89/// Deserializes a `Scopes` instance from a string representation.
90///
91/// The string representation is a semicolon-separated list of `key=value` pairs, following the
92/// Matrix URIs proposal. An empty scopes string (`"_"`) represents no scopes.
93fn deserialize_scopes<'de, D>(deserializer: D) -> Result<Scopes, D::Error>
94where
95    D: de::Deserializer<'de>,
96{
97    let s = Cow::<str>::deserialize(deserializer)?;
98    if s == EMPTY_SCOPES {
99        return Ok(Scopes::empty());
100    }
101
102    let scopes = s
103        .split(';')
104        .map(|s| {
105            let (key, value) = s
106                .split_once('=')
107                .ok_or_else(|| de::Error::custom("scope must be 'key=value'"))?;
108
109            Scope::create(key, value).map_err(de::Error::custom)
110        })
111        .collect::<Result<_, _>>()?;
112
113    Ok(scopes)
114}
115
116impl FromRequestParts<ServiceState> for Xt<ObjectContext> {
117    type Rejection = ObjectRejection;
118
119    async fn from_request_parts(
120        parts: &mut Parts,
121        state: &ServiceState,
122    ) -> Result<Self, Self::Rejection> {
123        let Path(params) = Path::<ContextParams>::from_request_parts(parts, state).await?;
124        let context = ObjectContext {
125            usecase: params.usecase,
126            scopes: params.scopes,
127        };
128
129        populate_sentry_context(&context);
130
131        let service = DownstreamService::from_request_parts(parts, state)
132            .await
133            .unwrap();
134
135        if state
136            .config
137            .killswitches
138            .matches(&context, service.as_str())
139        {
140            return Err(ObjectRejection::Killswitched);
141        }
142
143        state
144            .rate_limiter
145            .check(&context, None)
146            .map_err(ObjectRejection::RateLimited)?;
147
148        Ok(Xt(context))
149    }
150}
151
152/// Path parameters for extracting an [`ObjectContext`] from a request path.
153///
154/// Works on both collection-level (`/objects/{usecase}/{scopes}`) and object-level
155/// (`/objects/{usecase}/{scopes}/{*key}`) routes — the extra `key` parameter is ignored.
156#[derive(Clone, Debug, Deserialize)]
157pub(super) struct ContextParams {
158    pub usecase: Usecase,
159    #[serde(deserialize_with = "deserialize_scopes")]
160    pub scopes: Scopes,
161}
162
163fn populate_sentry_context(context: &ObjectContext) {
164    sentry::configure_scope(|s| {
165        s.set_tag("usecase", context.usecase.as_str());
166        for scope in &context.scopes {
167            s.set_tag(&format!("scope.{}", scope.name()), scope.value());
168        }
169    });
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175    use serde::de::IntoDeserializer;
176    use serde::de::value::{CowStrDeserializer, Error as DeError};
177    use std::borrow::Cow;
178
179    fn deser_scopes(input: &str) -> Result<Scopes, DeError> {
180        let deserializer: CowStrDeserializer<DeError> = Cow::Borrowed(input).into_deserializer();
181        deserialize_scopes(deserializer)
182    }
183
184    #[test]
185    fn parse_usecase() {
186        for (name, valid) in [
187            ("Debug_Files-123", true),
188            ("___", true),
189            ("_", false),
190            ("test/name", false),
191            ("test%2Fname", false),
192        ] {
193            let params = serde_json::json!({
194                "usecase": name,
195                "scopes": "org=123",
196                "key": "my-key",
197            });
198            assert_eq!(
199                serde_json::from_value::<ObjectParams>(params.clone()).is_ok(),
200                valid,
201                "object usecase {name:?}",
202            );
203            assert_eq!(
204                serde_json::from_value::<ContextParams>(params).is_ok(),
205                valid,
206                "context usecase {name:?}",
207            );
208        }
209    }
210
211    #[test]
212    fn parse_single_scope() {
213        let scopes = deser_scopes("org=123").unwrap();
214        assert_eq!(scopes.get_value("org"), Some("123"));
215    }
216
217    #[test]
218    fn parse_multiple_scopes() {
219        let scopes = deser_scopes("org=123;project=456").unwrap();
220        assert_eq!(scopes.get_value("org"), Some("123"));
221        assert_eq!(scopes.get_value("project"), Some("456"));
222    }
223
224    #[test]
225    fn parse_empty_scopes() {
226        let scopes = deser_scopes("_").unwrap();
227        assert!(scopes.is_empty());
228    }
229
230    #[test]
231    fn parse_missing_equals() {
232        let result = deser_scopes("org123");
233        assert!(result.is_err());
234    }
235
236    #[test]
237    fn parse_invalid_scope_chars() {
238        let result = deser_scopes("org=hello world");
239        assert!(result.is_err());
240    }
241
242    #[test]
243    fn parse_empty_key_or_value() {
244        assert!(deser_scopes("=value").is_err());
245        assert!(deser_scopes("key=").is_err());
246    }
247
248    // --- Extractor integration tests ---
249
250    use std::collections::BTreeMap;
251    use std::sync::Arc;
252
253    use axum::Router;
254    use axum::body::Body;
255    use axum::http::{Request, StatusCode};
256    use axum::routing::{get, post};
257    use objectstore_service::StorageService;
258    use objectstore_service::backend::in_memory::InMemoryBackend;
259    use objectstore_service::encryption::Cipher;
260    use tower::ServiceExt;
261
262    use crate::auth::PublicKeyDirectory;
263    use crate::config::Config;
264    use crate::killswitches::{Killswitch, Killswitches};
265    use crate::rate_limits::{RateLimiter, RateLimits, ThroughputLimits};
266    use crate::state::{ServiceState, Services};
267    use crate::web::RequestCounter;
268
269    async fn test_state(config: Config) -> ServiceState {
270        let service = StorageService::new(
271            Box::new(InMemoryBackend::new("in-memory")),
272            Cipher::ephemeral().unwrap(),
273        );
274        let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await.unwrap());
275        let rate_limiter = RateLimiter::new(config.rate_limits.clone());
276
277        Arc::new(Services {
278            config,
279            service,
280            key_directory,
281            rate_limiter,
282            request_counter: RequestCounter::new(0),
283        })
284    }
285
286    async fn handle_object_id(Xt(id): Xt<ObjectId>) -> String {
287        format!(
288            "usecase={} key={} scopes_empty={}",
289            id.context().usecase,
290            id.key(),
291            id.context().scopes.is_empty(),
292        )
293    }
294
295    async fn handle_object_context(Xt(ctx): Xt<ObjectContext>) -> String {
296        format!(
297            "usecase={} scopes_empty={}",
298            ctx.usecase,
299            ctx.scopes.is_empty(),
300        )
301    }
302
303    fn test_router(state: ServiceState) -> Router {
304        Router::new()
305            .route("/objects/{usecase}/{scopes}/{*key}", get(handle_object_id))
306            .route("/objects/{usecase}/{scopes}/", post(handle_object_context))
307            .with_state(state)
308    }
309
310    async fn response_body(response: http::Response<Body>) -> String {
311        let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
312            .await
313            .unwrap();
314        String::from_utf8(bytes.to_vec()).unwrap()
315    }
316
317    // Extraction tests
318
319    #[tokio::test]
320    async fn invalid_decoded_usecase_rejects_body_without_panicking() {
321        use crate::extractors::body::MeteredBody;
322
323        // Extract only the body so this exercises its own path rejection.
324        let app = Router::new()
325            .route(
326                "/objects/{usecase}/{scopes}/",
327                post(|_: MeteredBody| async { StatusCode::OK }),
328            )
329            .with_state(test_state(Config::default()).await);
330        let response = app
331            .oneshot(
332                Request::post("/objects/test%2Fname/org=123/")
333                    .body(Body::from("test data"))
334                    .unwrap(),
335            )
336            .await
337            .unwrap();
338        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
339        assert!(response_body(response).await.contains("usecase"));
340    }
341
342    #[tokio::test]
343    async fn extract_object_id_parses_path() {
344        let state = test_state(Config::default()).await;
345        let app = test_router(state);
346
347        let request = Request::builder()
348            .uri("/objects/myusecase/org=123;project=456/my-key")
349            .body(Body::empty())
350            .unwrap();
351        let response = app.oneshot(request).await.unwrap();
352
353        assert_eq!(response.status(), StatusCode::OK);
354        let body = response_body(response).await;
355        assert!(body.contains("usecase=myusecase"));
356        assert!(body.contains("key=my-key"));
357        assert!(body.contains("scopes_empty=false"));
358    }
359
360    #[tokio::test]
361    async fn extract_object_id_with_empty_scopes() {
362        let state = test_state(Config::default()).await;
363        let app = test_router(state);
364
365        let request = Request::builder()
366            .uri("/objects/myusecase/_/my-key")
367            .body(Body::empty())
368            .unwrap();
369        let response = app.oneshot(request).await.unwrap();
370
371        assert_eq!(response.status(), StatusCode::OK);
372        let body = response_body(response).await;
373        assert!(body.contains("scopes_empty=true"));
374    }
375
376    #[tokio::test]
377    async fn extract_object_context_parses_path() {
378        let state = test_state(Config::default()).await;
379        let app = test_router(state);
380
381        let request = Request::builder()
382            .method("POST")
383            .uri("/objects/myusecase/org=123;project=456/")
384            .body(Body::empty())
385            .unwrap();
386        let response = app.oneshot(request).await.unwrap();
387
388        assert_eq!(response.status(), StatusCode::OK);
389        let body = response_body(response).await;
390        assert!(body.contains("usecase=myusecase"));
391        assert!(body.contains("scopes_empty=false"));
392    }
393
394    #[tokio::test]
395    async fn extract_object_context_with_empty_scopes() {
396        let state = test_state(Config::default()).await;
397        let app = test_router(state);
398
399        let request = Request::builder()
400            .method("POST")
401            .uri("/objects/myusecase/_/")
402            .body(Body::empty())
403            .unwrap();
404        let response = app.oneshot(request).await.unwrap();
405
406        assert_eq!(response.status(), StatusCode::OK);
407        let body = response_body(response).await;
408        assert!(body.contains("scopes_empty=true"));
409    }
410
411    #[tokio::test]
412    async fn extract_object_id_invalid_scopes() {
413        let state = test_state(Config::default()).await;
414        let app = test_router(state);
415
416        let request = Request::builder()
417            .uri("/objects/myusecase/invalid-no-equals/key")
418            .body(Body::empty())
419            .unwrap();
420        let response = app.oneshot(request).await.unwrap();
421
422        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
423    }
424
425    // Killswitch tests
426
427    #[tokio::test]
428    async fn extract_object_id_killswitched() {
429        let config = Config {
430            killswitches: Killswitches::new(vec![Killswitch {
431                usecase: Some("blocked".parse().unwrap()),
432                scopes: BTreeMap::new(),
433                service: None,
434            }]),
435            ..Config::default()
436        };
437        let state = test_state(config).await;
438        let app = test_router(state);
439
440        let request = Request::builder()
441            .uri("/objects/blocked/org=1/key")
442            .body(Body::empty())
443            .unwrap();
444        let response = app.clone().oneshot(request).await.unwrap();
445        assert_eq!(response.status(), StatusCode::FORBIDDEN);
446
447        let request = Request::builder()
448            .uri("/objects/allowed/org=1/key")
449            .body(Body::empty())
450            .unwrap();
451        let response = app.oneshot(request).await.unwrap();
452        assert_eq!(response.status(), StatusCode::OK);
453    }
454
455    #[tokio::test]
456    async fn extract_object_context_killswitched() {
457        let config = Config {
458            killswitches: Killswitches::new(vec![Killswitch {
459                usecase: Some("blocked".parse().unwrap()),
460                scopes: BTreeMap::new(),
461                service: None,
462            }]),
463            ..Config::default()
464        };
465        let state = test_state(config).await;
466        let app = test_router(state);
467
468        let request = Request::builder()
469            .method("POST")
470            .uri("/objects/blocked/org=1/")
471            .body(Body::empty())
472            .unwrap();
473        let response = app.clone().oneshot(request).await.unwrap();
474        assert_eq!(response.status(), StatusCode::FORBIDDEN);
475
476        let request = Request::builder()
477            .method("POST")
478            .uri("/objects/allowed/org=1/")
479            .body(Body::empty())
480            .unwrap();
481        let response = app.oneshot(request).await.unwrap();
482        assert_eq!(response.status(), StatusCode::OK);
483    }
484
485    #[tokio::test]
486    async fn extract_object_id_killswitched_with_service() {
487        let config = Config {
488            killswitches: Killswitches::new(vec![Killswitch {
489                usecase: None,
490                scopes: BTreeMap::new(),
491                service: Some("test-*".into()),
492            }]),
493            ..Config::default()
494        };
495        let state = test_state(config).await;
496        let app = test_router(state);
497
498        // Matching service header → 403
499        let request = Request::builder()
500            .uri("/objects/any/org=1/key")
501            .header("x-downstream-service", "test-service")
502            .body(Body::empty())
503            .unwrap();
504        let response = app.clone().oneshot(request).await.unwrap();
505        assert_eq!(response.status(), StatusCode::FORBIDDEN);
506
507        // Non-matching service header → 200
508        let request = Request::builder()
509            .uri("/objects/any/org=1/key")
510            .header("x-downstream-service", "other-service")
511            .body(Body::empty())
512            .unwrap();
513        let response = app.clone().oneshot(request).await.unwrap();
514        assert_eq!(response.status(), StatusCode::OK);
515
516        // No service header → 200
517        let request = Request::builder()
518            .uri("/objects/any/org=1/key")
519            .body(Body::empty())
520            .unwrap();
521        let response = app.oneshot(request).await.unwrap();
522        assert_eq!(response.status(), StatusCode::OK);
523    }
524
525    // Rate limiter tests
526
527    #[tokio::test]
528    async fn extract_object_id_rate_limited() {
529        let config = Config {
530            rate_limits: RateLimits {
531                throughput: ThroughputLimits {
532                    global_rps: Some(1),
533                    burst: 0,
534                    ..ThroughputLimits::default()
535                },
536                ..RateLimits::default()
537            },
538            ..Config::default()
539        };
540        let state = test_state(config).await;
541        let app = test_router(state);
542
543        let request = Request::builder()
544            .uri("/objects/test/org=1/key")
545            .body(Body::empty())
546            .unwrap();
547        let response = app.clone().oneshot(request).await.unwrap();
548        assert_eq!(response.status(), StatusCode::OK);
549
550        let request = Request::builder()
551            .uri("/objects/test/org=1/key")
552            .body(Body::empty())
553            .unwrap();
554        let response = app.oneshot(request).await.unwrap();
555        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
556    }
557
558    #[tokio::test]
559    async fn extract_object_context_rate_limited() {
560        let config = Config {
561            rate_limits: RateLimits {
562                throughput: ThroughputLimits {
563                    global_rps: Some(1),
564                    burst: 0,
565                    ..ThroughputLimits::default()
566                },
567                ..RateLimits::default()
568            },
569            ..Config::default()
570        };
571        let state = test_state(config).await;
572        let app = test_router(state);
573
574        let request = Request::builder()
575            .method("POST")
576            .uri("/objects/test/org=1/")
577            .body(Body::empty())
578            .unwrap();
579        let response = app.clone().oneshot(request).await.unwrap();
580        assert_eq!(response.status(), StatusCode::OK);
581
582        let request = Request::builder()
583            .method("POST")
584            .uri("/objects/test/org=1/")
585            .body(Body::empty())
586            .unwrap();
587        let response = app.oneshot(request).await.unwrap();
588        assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
589    }
590}
objectstore_server/extractors/id.rs

objectstore_server/extractors/
id.rs
