secretspec_ipc/
provider.rs

1use crate::error::{ErrorKind, RpcError};
2use crate::protocol::PROVIDER_PROTOCOL;
3use crate::protocol::callback::{self, CredentialParams, CredentialResult};
4use crate::protocol::provider::{
5    Address, AddressParams, CAPABILITIES, ClearParams, ClearResult, DescribeWriteTargetResult,
6    ExistsResult, GetManyParams, GetManyResult, GetResult, InitializeApplication,
7    InitializedApplication, Metadata, ReflectParams, ReflectResult, ResolveAddressResult,
8    SetExpiringParams, SetParams, method,
9};
10use crate::server::{ApplicationHandler, RequestContext, RpcResult, ServerConfig};
11use async_trait::async_trait;
12use serde::de::DeserializeOwned;
13use serde_json::{Value, json};
14use std::sync::Arc;
15use std::sync::OnceLock;
16use zeroize::Zeroizing;
17
18/// Owned value passed across the endpoint-author boundary. Its backing bytes
19/// are cleared on drop independently of the JSON frame buffer.
20#[derive(Clone)]
21pub struct SecretValue(Zeroizing<String>);
22
23impl std::fmt::Debug for SecretValue {
24    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25        formatter
26            .debug_tuple("SecretValue")
27            .field(&crate::protocol::Redacted)
28            .finish()
29    }
30}
31
32impl SecretValue {
33    pub fn new(value: String) -> Self {
34        Self(Zeroizing::new(value))
35    }
36
37    pub fn expose(&self) -> &str {
38        self.0.as_str()
39    }
40
41    /// Copies the value out for JSON serialization.
42    ///
43    /// Deliberately a copy rather than `mem::take`: taking would move the
44    /// backing allocation out of the `Zeroizing` wrapper, leaving it to wipe an
45    /// empty string while the real bytes lived on un-wiped. Copying keeps this
46    /// value's own buffer covered by its destructor.
47    fn into_string(self) -> String {
48        self.0.as_str().to_owned()
49    }
50}
51
52/// A provider value and the absolute time at which the value itself expires.
53///
54/// `None` means the provider does not know a validity bound; it does not mean
55/// the value is permanent. Cache freshness is owned by the resolver and is not
56/// represented here.
57#[derive(Debug, Clone)]
58pub struct ProvidedSecret {
59    value: SecretValue,
60    expires_at_unix_ms: Option<u64>,
61    revision: Option<crate::Revision>,
62}
63
64impl ProvidedSecret {
65    pub fn new(value: String, expires_at_unix_ms: Option<u64>) -> Self {
66        Self {
67            value: SecretValue::new(value),
68            expires_at_unix_ms,
69            revision: None,
70        }
71    }
72
73    pub fn value(&self) -> &str {
74        self.value.expose()
75    }
76
77    pub const fn expires_at_unix_ms(&self) -> Option<u64> {
78        self.expires_at_unix_ms
79    }
80
81    /// Attach non-secret identity/version metadata from the same read (0.21+).
82    pub fn with_revision(mut self, revision: Option<crate::Revision>) -> Self {
83        self.revision = revision;
84        self
85    }
86
87    pub fn revision(&self) -> Option<&crate::Revision> {
88        self.revision.as_ref()
89    }
90
91    fn into_parts(self) -> (String, Option<u64>, Option<crate::Revision>) {
92        (
93            self.value.into_string(),
94            self.expires_at_unix_ms,
95            self.revision,
96        )
97    }
98}
99
100#[async_trait]
101pub trait ProviderHandler: Send + Sync + 'static {
102    /// Supported operation names. `provider.resolve_address` is mandatory.
103    fn capabilities(&self) -> Vec<String>;
104
105    async fn initialize(
106        &self,
107        context: &RequestContext,
108        application: InitializeApplication,
109    ) -> RpcResult<Metadata>;
110
111    async fn resolve_address(
112        &self,
113        context: RequestContext,
114        address: Address,
115    ) -> RpcResult<ResolveAddressResult>;
116
117    async fn get(
118        &self,
119        _context: RequestContext,
120        _address: Address,
121    ) -> RpcResult<Option<ProvidedSecret>> {
122        Err(RpcError::new(ErrorKind::CapabilityRequired))
123    }
124
125    async fn get_many(
126        &self,
127        _context: RequestContext,
128        _params: GetManyParams,
129    ) -> RpcResult<GetManyResult> {
130        Err(RpcError::new(ErrorKind::CapabilityRequired))
131    }
132
133    async fn exists(&self, _context: RequestContext, _address: Address) -> RpcResult<bool> {
134        Err(RpcError::new(ErrorKind::CapabilityRequired))
135    }
136
137    async fn set(
138        &self,
139        _context: RequestContext,
140        _address: Address,
141        _value: SecretValue,
142    ) -> RpcResult<()> {
143        Err(RpcError::new(ErrorKind::CapabilityRequired))
144    }
145
146    async fn set_expiring(
147        &self,
148        _context: RequestContext,
149        _address: Address,
150        _value: SecretValue,
151        _ttl_ms: u64,
152    ) -> RpcResult<()> {
153        Err(RpcError::new(ErrorKind::CapabilityRequired))
154    }
155
156    async fn delete(&self, _context: RequestContext, _address: Address) -> RpcResult<bool> {
157        Err(RpcError::new(ErrorKind::CapabilityRequired))
158    }
159
160    async fn clear(&self, _context: RequestContext, _params: ClearParams) -> RpcResult<usize> {
161        Err(RpcError::new(ErrorKind::CapabilityRequired))
162    }
163
164    async fn check_writable(&self, _context: RequestContext, _address: Address) -> RpcResult<()> {
165        Err(RpcError::new(ErrorKind::CapabilityRequired))
166    }
167
168    async fn check_deletable(&self, _context: RequestContext, _address: Address) -> RpcResult<()> {
169        Err(RpcError::new(ErrorKind::CapabilityRequired))
170    }
171
172    async fn describe_write_target(
173        &self,
174        _context: RequestContext,
175        _address: Address,
176    ) -> RpcResult<String> {
177        Err(RpcError::new(ErrorKind::CapabilityRequired))
178    }
179
180    async fn reflect(
181        &self,
182        _context: RequestContext,
183        _params: ReflectParams,
184    ) -> RpcResult<ReflectResult> {
185        Err(RpcError::new(ErrorKind::CapabilityRequired))
186    }
187
188    async fn shutdown(&self) {}
189}
190
191/// Requests one provider credential from the client while serving the current
192/// request, normally from [`ProviderHandler::initialize`] (0.21+).
193///
194/// A client that did not advertise credential brokerage behaves like an empty
195/// broker, allowing an endpoint to retain native environment, agent, or
196/// workload-identity fallbacks without treating callback support as mandatory.
197pub async fn request_credential(
198    context: &RequestContext,
199    params: CredentialParams,
200) -> RpcResult<Option<SecretValue>> {
201    params
202        .validate()
203        .map_err(|_| RpcError::new(ErrorKind::InvalidParams))?;
204    if !context.peer.supports(callback::method::CREDENTIAL) {
205        return Ok(None);
206    }
207    let result: CredentialResult = context
208        .peer
209        .call(callback::method::CREDENTIAL, &params, context)
210        .await?;
211    result
212        .validate()
213        .map_err(|_| RpcError::new(ErrorKind::OperationFailed))?;
214    Ok(match result {
215        CredentialResult::Found { value } => Some(SecretValue::new(value)),
216        CredentialResult::Missing => None,
217    })
218}
219
220struct ProviderApplication<H> {
221    handler: Arc<H>,
222    metadata: OnceLock<Metadata>,
223}
224
225impl<H> ProviderApplication<H> {
226    fn new(handler: Arc<H>) -> Self {
227        Self {
228            handler,
229            metadata: OnceLock::new(),
230        }
231    }
232
233    fn validate_address(&self, address: &Address) -> RpcResult<()> {
234        address.validate().map_err(invalid_params)?;
235        let metadata = self
236            .metadata
237            .get()
238            .ok_or_else(|| RpcError::new(ErrorKind::Internal))?;
239        if let Address::Native { coordinates } = address
240            && coordinates
241                .unsupported(&metadata.supported_coordinates)
242                .is_some()
243        {
244            return Err(RpcError::new(ErrorKind::InvalidParams));
245        }
246        Ok(())
247    }
248
249    fn address_params(&self, value: Value) -> RpcResult<AddressParams> {
250        let params: AddressParams = parse(value)?;
251        self.validate_address(&params.address)?;
252        Ok(params)
253    }
254}
255
256#[async_trait]
257impl<H: ProviderHandler> ApplicationHandler for ProviderApplication<H> {
258    fn protocol(&self) -> &'static str {
259        PROVIDER_PROTOCOL
260    }
261
262    fn capabilities(&self) -> Vec<String> {
263        let mut capabilities = self.handler.capabilities();
264        capabilities.retain(|capability| CAPABILITIES.contains(&capability.as_str()));
265        capabilities.sort();
266        capabilities.dedup();
267        capabilities
268    }
269
270    fn validate_capabilities(&self, capabilities: &[String]) -> RpcResult<()> {
271        crate::protocol::provider::validate_capabilities(capabilities)
272            .map_err(|_| RpcError::new(ErrorKind::CapabilityRequired))
273    }
274
275    async fn initialize(&self, context: &RequestContext, application: Value) -> RpcResult<Value> {
276        let application: InitializeApplication = parse(application)?;
277        application.validate().map_err(invalid_params)?;
278        let scheme = application.scheme.clone();
279        let metadata = self.handler.initialize(context, application).await?;
280        metadata
281            .validate()
282            .map_err(|_| RpcError::new(ErrorKind::OperationFailed))?;
283        if metadata.name != scheme {
284            return Err(RpcError::new(ErrorKind::Conflict));
285        }
286        self.metadata
287            .set(metadata.clone())
288            .map_err(|_| RpcError::new(ErrorKind::Conflict))?;
289        serde_json::to_value(InitializedApplication { provider: metadata })
290            .map_err(|_| RpcError::new(ErrorKind::Internal))
291    }
292
293    async fn call(&self, context: RequestContext, method: &str, params: Value) -> RpcResult<Value> {
294        match method {
295            method::RESOLVE_ADDRESS => {
296                let params = self.address_params(params)?;
297                let result = self
298                    .handler
299                    .resolve_address(context, params.address)
300                    .await?;
301                result
302                    .coordinates
303                    .validate()
304                    .map_err(|_| RpcError::new(ErrorKind::OperationFailed))?;
305                if result
306                    .coordinates
307                    .unsupported(
308                        &self
309                            .metadata
310                            .get()
311                            .expect("provider initialized before calls")
312                            .supported_coordinates,
313                    )
314                    .is_some()
315                {
316                    return Err(RpcError::new(ErrorKind::OperationFailed));
317                }
318                encode(result)
319            }
320            method::GET => {
321                let params = self.address_params(params)?;
322                let result = match self.handler.get(context, params.address).await? {
323                    Some(value) => {
324                        let (value, expires_at_unix_ms, revision) = value.into_parts();
325                        GetResult::Found {
326                            value,
327                            expires_at_unix_ms,
328                            revision,
329                        }
330                    }
331                    None => GetResult::Missing,
332                };
333                encode(result)
334            }
335            method::GET_MANY => {
336                let params: GetManyParams = parse(params)?;
337                params.validate().map_err(invalid_params)?;
338                for request in &params.requests {
339                    self.validate_address(&request.address)?;
340                }
341                let expected = params
342                    .requests
343                    .iter()
344                    .map(|request| request.name.clone())
345                    .collect::<Vec<_>>();
346                let result = self.handler.get_many(context, params).await?;
347                if result.results.len() != expected.len()
348                    || result
349                        .results
350                        .iter()
351                        .zip(expected)
352                        .any(|(result, expected)| result.name != expected)
353                {
354                    return Err(RpcError::new(ErrorKind::OperationFailed));
355                }
356                encode(result)
357            }
358            method::EXISTS => {
359                let params = self.address_params(params)?;
360                encode(ExistsResult {
361                    exists: self.handler.exists(context, params.address).await?,
362                })
363            }
364            method::SET => {
365                let params: SetParams = parse(params)?;
366                params.validate().map_err(invalid_params)?;
367                self.validate_address(&params.address)?;
368                self.handler
369                    .set(context, params.address, SecretValue::new(params.value))
370                    .await?;
371                Ok(json!({"stored": true}))
372            }
373            method::SET_EXPIRING => {
374                let params: SetExpiringParams = parse(params)?;
375                params.validate().map_err(invalid_params)?;
376                self.validate_address(&params.address)?;
377                self.handler
378                    .set_expiring(
379                        context,
380                        params.address,
381                        SecretValue::new(params.value),
382                        params.ttl_ms,
383                    )
384                    .await?;
385                Ok(json!({"stored": true}))
386            }
387            method::DELETE => {
388                let params = self.address_params(params)?;
389                Ok(json!({"deleted": self.handler.delete(context, params.address).await?}))
390            }
391            method::CLEAR => {
392                let params: ClearParams = parse(params)?;
393                params.validate().map_err(invalid_params)?;
394                let cleared = self.handler.clear(context, params).await?;
395                encode(ClearResult { cleared })
396            }
397            method::CHECK_WRITABLE => {
398                let params = self.address_params(params)?;
399                self.handler.check_writable(context, params.address).await?;
400                Ok(json!({}))
401            }
402            method::CHECK_DELETABLE => {
403                let params = self.address_params(params)?;
404                self.handler
405                    .check_deletable(context, params.address)
406                    .await?;
407                Ok(json!({}))
408            }
409            method::DESCRIBE_WRITE_TARGET => {
410                let params = self.address_params(params)?;
411                encode(DescribeWriteTargetResult {
412                    description: self
413                        .handler
414                        .describe_write_target(context, params.address)
415                        .await?,
416                })
417            }
418            method::REFLECT => {
419                let params: ReflectParams = parse(params)?;
420                params.validate().map_err(invalid_params)?;
421                let result = self.handler.reflect(context, params).await?;
422                result
423                    .validate()
424                    .map_err(|_| RpcError::new(ErrorKind::OperationFailed))?;
425                encode(result)
426            }
427            _ => Err(RpcError::new(ErrorKind::MethodNotFound)),
428        }
429    }
430
431    async fn shutdown(&self) {
432        self.handler.shutdown().await;
433    }
434}
435
436fn parse<T: DeserializeOwned>(value: Value) -> RpcResult<T> {
437    serde_json::from_value(value).map_err(|_| RpcError::new(ErrorKind::InvalidParams))
438}
439
440fn encode<T: serde::Serialize>(value: T) -> RpcResult<Value> {
441    serde_json::to_value(value).map_err(|_| RpcError::new(ErrorKind::Internal))
442}
443
444fn invalid_params(_: crate::Error) -> RpcError {
445    RpcError::new(ErrorKind::InvalidParams)
446}
447
448/// Serve one typed provider endpoint without assembling the generic adapter.
449pub async fn serve_provider<R, W, H>(
450    reader: R,
451    writer: W,
452    handler: H,
453    config: ServerConfig,
454) -> crate::Result<()>
455where
456    R: tokio::io::AsyncRead + Unpin + Send + 'static,
457    W: tokio::io::AsyncWrite + Unpin + Send + 'static,
458    H: ProviderHandler,
459{
460    crate::server::serve(
461        reader,
462        writer,
463        Arc::new(ProviderApplication::new(Arc::new(handler))),
464        config,
465    )
466    .await
467}
secretspec_ipc/provider.rs

secretspec_ipc/
provider.rs
