unitycatalog_storage_proxy/
client.rs

1//! The portable **client arm**: a [`StorageProxyBackend`] backed by a
2//! [`UnityObjectStoreFactory`].
3//!
4//! Given only `{baseUrl, token}` this arm serves the proxy against **any** Unity
5//! Catalog server — the factory resolves the securable and vends a scoped
6//! credential through the UC REST API, so there is zero coupling to any particular
7//! server implementation. Authorization is delegated to the upstream UC server
8//! (a vend for an unauthorized securable fails), plus the local traversal guard.
9//!
10//! Requires the `client-arm` feature.
11
12use std::sync::Arc;
13
14use object_store::DynObjectStore;
15use unitycatalog_object_store::{
16    PathOperation, TableOperation, UnityObjectStoreFactory, VolumeOperation,
17};
18
19use crate::backend::{
20    ForwardedUser, ProxyCapabilities, ProxyReq, Securable, StorageProxyBackend,
21    reject_key_traversal,
22};
23use crate::error::{ProxyError, ProxyResult};
24
25/// Default header the client arm re-emits the forwarded end-user identity under
26/// on upstream Unity Catalog requests. Matches the reverse-proxy auth layer's
27/// [`DEFAULT_FORWARDED_USER_HEADER`](crate::auth::DEFAULT_FORWARDED_USER_HEADER)
28/// so the incoming and outgoing header names line up by default. Duplicated here
29/// (as a literal) so the client arm does not depend on the `bin`-gated `auth`
30/// module.
31pub const DEFAULT_FORWARDED_USER_HEADER: &str = "x-forwarded-user";
32
33/// A [`StorageProxyBackend`] that resolves + vends through a
34/// [`UnityObjectStoreFactory`]. Portable against any UC server.
35#[derive(Clone)]
36pub struct UnityFactoryProxyBackend {
37    factory: UnityObjectStoreFactory,
38    /// Header name the forwarded end-user identity (from the request context) is
39    /// re-emitted under on upstream UC calls. Defaults to
40    /// [`DEFAULT_FORWARDED_USER_HEADER`].
41    forwarded_header: String,
42}
43
44impl UnityFactoryProxyBackend {
45    /// Connect to a Unity Catalog server at `base_uri` (e.g.
46    /// `https://host/api/2.1/unity-catalog/`) with an optional bearer `token`.
47    ///
48    /// Forwards a per-request end-user identity to upstream UC under
49    /// [`DEFAULT_FORWARDED_USER_HEADER`]. Use
50    /// [`connect_with_forwarded_header`](Self::connect_with_forwarded_header) to
51    /// re-emit the identity under a different header name.
52    pub async fn connect(base_uri: impl Into<String>, token: Option<String>) -> ProxyResult<Self> {
53        Self::connect_with_forwarded_header(base_uri, token, DEFAULT_FORWARDED_USER_HEADER).await
54    }
55
56    /// Like [`connect`](Self::connect), but re-emits the forwarded end-user
57    /// identity under `forwarded_header` on upstream UC requests (typically the
58    /// same header name the proxy read the incoming identity from).
59    pub async fn connect_with_forwarded_header(
60        base_uri: impl Into<String>,
61        token: Option<String>,
62        forwarded_header: impl Into<String>,
63    ) -> ProxyResult<Self> {
64        let allow_unauthenticated = token.is_none();
65        let factory = UnityObjectStoreFactory::builder()
66            .with_uri(base_uri)
67            .with_token(token)
68            .with_allow_unauthenticated(allow_unauthenticated)
69            .build()
70            .await
71            .map_err(|e| ProxyError::Internal(format!("connect UC factory: {e}")))?;
72        Ok(Self {
73            factory,
74            forwarded_header: forwarded_header.into(),
75        })
76    }
77
78    /// Build from an already-constructed factory (e.g. one sharing a client pool).
79    /// Forwards identity under [`DEFAULT_FORWARDED_USER_HEADER`].
80    pub fn from_factory(factory: UnityObjectStoreFactory) -> Self {
81        Self {
82            factory,
83            forwarded_header: DEFAULT_FORWARDED_USER_HEADER.to_string(),
84        }
85    }
86}
87
88#[async_trait::async_trait]
89impl<Cx: ForwardedUser + Send + Sync + 'static> StorageProxyBackend<Cx>
90    for UnityFactoryProxyBackend
91{
92    fn capabilities(&self) -> ProxyCapabilities {
93        ProxyCapabilities { enabled: true }
94    }
95
96    async fn authorize(&self, req: &ProxyReq, _cx: &Cx) -> ProxyResult<()> {
97        // The confused-deputy prefix confinement is intrinsic here: the vend
98        // returns a credential scoped to the securable root, and `open` returns a
99        // store prefixed there, so a key cannot address a sibling. We add the
100        // traversal guard (layer 1) and defer permission checks to the vend itself.
101        reject_key_traversal(&req.key)
102    }
103
104    async fn open(&self, req: &ProxyReq, cx: &Cx) -> ProxyResult<Arc<DynObjectStore>> {
105        // Forward the validated end-user identity (if any) to the upstream UC
106        // vend + metadata calls, so UC attributes the credential to the real user
107        // rather than to the proxy's own service principal. An anonymous request
108        // (`None`) reuses the base factory unchanged (its static token / auth).
109        let factory = self
110            .factory
111            .with_forwarded_user(&self.forwarded_header, cx.forwarded_user())
112            .map_err(ProxyError::Storage)?;
113
114        let write = req.verb.is_write();
115        let store = match &req.securable {
116            Securable::Table { full_name } => {
117                let op = if write {
118                    TableOperation::ReadWrite
119                } else {
120                    TableOperation::Read
121                };
122                factory.for_table(full_name.clone(), op).await
123            }
124            Securable::Volume { full_name } => {
125                let op = if write {
126                    VolumeOperation::ReadWrite
127                } else {
128                    VolumeOperation::Read
129                };
130                factory.for_volume(full_name.clone(), op).await
131            }
132            Securable::Path { url } => {
133                let op = if write {
134                    PathOperation::ReadWrite
135                } else {
136                    PathOperation::Read
137                };
138                factory.for_path(url, op).await
139            }
140        }
141        .map_err(ProxyError::Storage)?;
142
143        // `as_dyn()` returns the store prefixed at the credential-scoped root, so
144        // `req.key` addresses relative to the securable.
145        Ok(store.as_dyn())
146    }
147}
unitycatalog_storage_proxy/client.rs

unitycatalog_storage_proxy/
client.rs
