bitwarden_access_connector/api/
mod.rs

1//! HTTP API client wrappers for the Bitwarden server's PAM rotation endpoints.
2//!
3//! [`build_sdk_client`] assembles the client the connector runs on; [`AccessConnectorClient`]
4//! reaches through its configurations, mapping calls into domain types.
5
6pub(crate) mod models;
7
8use std::{sync::Arc, time::Duration};
9
10use bitwarden_api_api::{
11    apis::AuthRequired,
12    models::{
13        CipherMiniResponseModel, CipherRequestModel, ReportRotationFailedRequestModel,
14        ReportRotationSucceededRequestModel,
15    },
16};
17use bitwarden_core::{
18    Client, ClientBuilder, ClientSettings, FromClient, client::internal::ApiConfigurations,
19    key_management::KeySlotIds,
20};
21use bitwarden_crypto::KeyStore;
22use bitwarden_pam::{
23    RotationFailureCode, RotationSyncState, SessionTerminationOutcome, WorkSnapshot,
24};
25use bitwarden_vault::Cipher;
26use models::{ApiError, JobRef};
27use reqwest_middleware::{Middleware, Next};
28use tokio::{sync::watch, time::Instant};
29use uuid::Uuid;
30
31use crate::{
32    auth::{
33        session::{SessionError, SessionManager},
34        token_handler::AccessConnectorTokenHandler,
35    },
36    error::SafeDetail,
37};
38
39/// [`reqwest_middleware::Middleware`] that attaches the connector bearer token and retries one 401
40/// after a forced refresh. Unlike bitwarden-auth's middleware, it fails on session loss instead of
41/// sending the request without a token.
42pub(crate) struct AccessConnectorAuthMiddleware {
43    handler: AccessConnectorTokenHandler,
44}
45
46impl AccessConnectorAuthMiddleware {
47    pub(crate) fn new(handler: AccessConnectorTokenHandler) -> Self {
48        Self { handler }
49    }
50
51    /// Core builds this middleware before startup has a session; a request in that window fails
52    /// rather than going out unauthenticated.
53    fn session(&self) -> Result<Arc<SessionManager>, reqwest_middleware::Error> {
54        self.handler.session().ok_or_else(|| {
55            tracing::error!("authenticated request issued before the session was installed");
56            reqwest_middleware::Error::Middleware(anyhow::Error::new(SessionNotInstalledError))
57        })
58    }
59
60    async fn get_bearer(&self) -> Result<String, reqwest_middleware::Error> {
61        self.session()?
62            .bearer(None)
63            .await
64            .map_err(|e| reqwest_middleware::Error::Middleware(anyhow::Error::new(e)))
65    }
66
67    async fn force_refresh_bearer(&self, stale: &str) -> Result<String, reqwest_middleware::Error> {
68        self.session()?
69            .force_refresh(stale, None)
70            .await
71            .map_err(|e| reqwest_middleware::Error::Middleware(anyhow::Error::new(e)))
72    }
73}
74
75/// A request needed a bearer before [`AccessConnectorTokenHandler::set_session`] ran.
76#[derive(Debug, thiserror::Error)]
77#[error("session not installed")]
78pub(crate) struct SessionNotInstalledError;
79
80#[async_trait::async_trait]
81impl Middleware for AccessConnectorAuthMiddleware {
82    async fn handle(
83        &self,
84        mut req: reqwest::Request,
85        ext: &mut http::Extensions,
86        next: Next<'_>,
87    ) -> Result<reqwest::Response, reqwest_middleware::Error> {
88        // The generated API opts into auth via the AuthRequired::Bearer extension.
89        let auth_required = matches!(ext.get::<AuthRequired>(), Some(AuthRequired::Bearer));
90
91        let used_token: Option<String> = if auth_required {
92            let token = self.get_bearer().await?;
93            attach_bearer_header(&mut req, &token);
94            Some(token)
95        } else {
96            None
97        };
98
99        // Clone before `run` consumes the request; a streaming body cannot be cloned and gets no
100        // retry.
101        let req_clone = req.try_clone();
102
103        let response = next.clone().run(req, ext).await?;
104
105        if auth_required
106            && let Some(mut cloned) = req_clone
107            && response.status() == http::StatusCode::UNAUTHORIZED
108        {
109            tracing::info!("connector API: 401 received, refreshing token and retrying");
110
111            let stale = used_token.as_deref().unwrap_or("");
112            let new_token = self.force_refresh_bearer(stale).await?;
113            attach_bearer_header(&mut cloned, &new_token);
114
115            return next.run(cloned, ext).await;
116        }
117
118        Ok(response)
119    }
120}
121
122fn attach_bearer_header(req: &mut reqwest::Request, token: &str) {
123    let value = match format!("Bearer {token}").parse::<http::HeaderValue>() {
124        Ok(v) => v,
125        Err(e) => {
126            // The token has a character invalid in a header value. Proceed without it; the
127            // server's 401 surfaces the error through the retry path.
128            tracing::warn!("connector API: cannot format bearer token as header value: {e}");
129            return;
130        }
131    };
132    req.headers_mut().insert(http::header::AUTHORIZATION, value);
133}
134
135/// Bounds each request so a black-holed connection cannot stall the connector.
136pub(crate) const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
137
138/// Applies [`REQUEST_TIMEOUT`] to a whole request; core sets none.
139pub(crate) struct TimeoutMiddleware {
140    timeout: Duration,
141}
142
143impl TimeoutMiddleware {
144    pub(crate) fn new(timeout: Duration) -> Self {
145        Self { timeout }
146    }
147}
148
149#[async_trait::async_trait]
150impl Middleware for TimeoutMiddleware {
151    async fn handle(
152        &self,
153        req: reqwest::Request,
154        ext: &mut http::Extensions,
155        next: Next<'_>,
156    ) -> Result<reqwest::Response, reqwest_middleware::Error> {
157        tokio::time::timeout(self.timeout, next.run(req, ext))
158            .await
159            .unwrap_or_else(|_| {
160                Err(reqwest_middleware::Error::Middleware(anyhow::Error::new(
161                    RequestTimedOutError,
162                )))
163            })
164    }
165}
166
167/// A request exceeded [`REQUEST_TIMEOUT`].
168#[derive(Debug, thiserror::Error)]
169#[error("request timed out")]
170pub(crate) struct RequestTimedOutError;
171
172/// Records the last successful server contact, so target-side steps can pause while the server is
173/// unreachable. Any 2xx counts, even if the body later fails to parse.
174pub(crate) struct ConnectivityMiddleware {
175    last_ok_tx: watch::Sender<Instant>,
176}
177
178impl ConnectivityMiddleware {
179    pub(crate) fn new(last_ok_tx: watch::Sender<Instant>) -> Self {
180        Self { last_ok_tx }
181    }
182}
183
184#[async_trait::async_trait]
185impl Middleware for ConnectivityMiddleware {
186    async fn handle(
187        &self,
188        req: reqwest::Request,
189        ext: &mut http::Extensions,
190        next: Next<'_>,
191    ) -> Result<reqwest::Response, reqwest_middleware::Error> {
192        let response = next.run(req, ext).await?;
193        if response.status().is_success() {
194            self.last_ok_tx.send_modify(|t| *t = Instant::now());
195        }
196        Ok(response)
197    }
198}
199
200/// Builds the SDK client the connector runs against, plus the handler that carries its session.
201///
202/// Core only disables redirect-following when extra middleware is supplied, so the middleware vec
203/// must stay non-empty. These endpoints never redirect; following one would treat another host's
204/// reply as the API's and, on a 307, forward the encrypted cipher body.
205///
206/// The timeout is outermost, so it also bounds token renewal. [`ConnectivityMiddleware`] sits
207/// outside the auth middleware, so a 401 retried into a 2xx counts once.
208pub(crate) fn build_sdk_client(
209    api_url: String,
210    identity_url: String,
211    timeout: Duration,
212    last_ok_tx: watch::Sender<Instant>,
213) -> (Client, AccessConnectorTokenHandler) {
214    let handler = AccessConnectorTokenHandler::default();
215
216    let client = ClientBuilder::new()
217        .with_settings(ClientSettings {
218            api_url,
219            identity_url,
220            ..Default::default()
221        })
222        .with_token_handler(Arc::new(handler.clone()))
223        .with_middleware(vec![
224            Arc::new(TimeoutMiddleware::new(timeout)),
225            Arc::new(ConnectivityMiddleware::new(last_ok_tx)),
226        ])
227        .build();
228
229    (client, handler)
230}
231
232/// Domain-typed wrapper around the generated PAM rotation API clients.
233#[derive(FromClient)]
234pub(crate) struct AccessConnectorClient {
235    pub(crate) key_store: KeyStore<KeySlotIds>,
236    pub(crate) api_configurations: Arc<ApiConfigurations>,
237}
238
239/// Extension trait that exposes [`AccessConnectorClient`] on [`Client`].
240pub(crate) trait AccessConnectorClientExt {
241    fn access_connector(&self) -> AccessConnectorClient;
242}
243
244impl AccessConnectorClientExt for Client {
245    fn access_connector(&self) -> AccessConnectorClient {
246        AccessConnectorClient::from_client(self)
247    }
248}
249
250impl AccessConnectorClient {
251    /// Poll for claimable rotation jobs. A 404 maps to [`ApiError::NotEligible`].
252    pub(crate) async fn poll_jobs(&self) -> Result<Vec<JobRef>, ApiError> {
253        let result = self
254            .api_configurations
255            .api_client
256            .pam_access_connector_rotation_jobs_api()
257            .get_all()
258            .await;
259
260        match result {
261            Ok(list_model) => {
262                let jobs = list_model
263                    .data
264                    .unwrap_or_default()
265                    .into_iter()
266                    .filter_map(|item| item.job_id.map(|id| JobRef { id }))
267                    .collect();
268                Ok(jobs)
269            }
270            Err(e) => Err(classify_error(e, Route::AccessConnectorOrJob)),
271        }
272    }
273
274    /// Attempt to claim a rotation job.
275    ///
276    /// `Ok(None)` means another connector won the race (409). A 404 maps to
277    /// [`ApiError::NotEligible`].
278    pub(crate) async fn claim(&self, job_id: Uuid) -> Result<Option<WorkSnapshot>, ApiError> {
279        let result = self
280            .api_configurations
281            .api_client
282            .pam_access_connector_rotation_jobs_api()
283            .claim(job_id)
284            .await;
285
286        match result {
287            Ok(model) => {
288                let snapshot =
289                    WorkSnapshot::try_from(model).map_err(|e| ApiError::Protocol(e.to_string()))?;
290                Ok(Some(snapshot))
291            }
292            Err(bitwarden_api_base::Error::Response(ref rc)) if rc.status.as_u16() == 409 => {
293                Ok(None)
294            }
295            Err(e) => Err(classify_error(e, Route::AccessConnectorOrJob)),
296        }
297    }
298
299    /// Fetch the encrypted cipher for an attempt. A 404 maps to [`ApiError::UnknownAttempt`].
300    pub(crate) async fn get_cipher(&self, attempt_id: Uuid) -> Result<Cipher, ApiError> {
301        let result = self
302            .api_configurations
303            .api_client
304            .pam_access_connector_rotation_attempts_api()
305            .get_cipher(attempt_id)
306            .await;
307
308        match result {
309            Ok(model) => parse_cipher(model),
310            Err(e) => Err(classify_error(e, Route::Attempt)),
311        }
312    }
313
314    /// Write the re-encrypted cipher back. `lastKnownRevisionDate` is the optimistic-concurrency
315    /// check; a 409 means revision drift or capability lost, a 404 an unknown attempt.
316    pub(crate) async fn put_cipher(
317        &self,
318        attempt_id: Uuid,
319        body: CipherRequestModel,
320    ) -> Result<(), ApiError> {
321        let result = self
322            .api_configurations
323            .api_client
324            .pam_access_connector_rotation_attempts_api()
325            .put_cipher(attempt_id, body)
326            .await;
327
328        match result {
329            Ok(()) => Ok(()),
330            Err(e) => Err(classify_error(e, Route::Attempt)),
331        }
332    }
333
334    /// Report a successful rotation attempt. A 409 or 404 is final: do not retry.
335    pub(crate) async fn report_success(
336        &self,
337        attempt_id: Uuid,
338        termination: SessionTerminationOutcome,
339    ) -> Result<(), ApiError> {
340        let body = ReportRotationSucceededRequestModel {
341            session_termination: termination.try_into().map_err(unrecognized_variant)?,
342        };
343
344        let result = self
345            .api_configurations
346            .api_client
347            .pam_access_connector_rotation_attempts_api()
348            .success(attempt_id, body)
349            .await;
350
351        match result {
352            Ok(()) => Ok(()),
353            Err(e) => Err(classify_error(e, Route::Attempt)),
354        }
355    }
356
357    /// Report a failed rotation attempt. A 409 or 404 is final, as in [`Self::report_success`].
358    pub(crate) async fn report_failure(
359        &self,
360        attempt_id: Uuid,
361        code: RotationFailureCode,
362        detail: Option<SafeDetail>,
363        sync_state: RotationSyncState,
364    ) -> Result<(), ApiError> {
365        let body = ReportRotationFailedRequestModel {
366            sync_state: sync_state.try_into().map_err(unrecognized_variant)?,
367            error_code: code.as_token().map_err(unrecognized_variant)?.to_owned(),
368            detail: detail.map(|d| d.as_str().to_owned()),
369        };
370
371        let result = self
372            .api_configurations
373            .api_client
374            .pam_access_connector_rotation_attempts_api()
375            .failure(attempt_id, body)
376            .await;
377
378        match result {
379            Ok(()) => Ok(()),
380            Err(e) => Err(classify_error(e, Route::Attempt)),
381        }
382    }
383}
384
385/// Route class, for disambiguating 404 semantics.
386#[derive(Clone, Copy)]
387enum Route {
388    /// `jobs` and `jobs/{id}/claim`: a 404 means the connector is not eligible.
389    AccessConnectorOrJob,
390    /// `attempts/{id}/…`: a 404 means the server does not know the attempt.
391    Attempt,
392}
393
394fn classify_error(err: bitwarden_api_base::Error, route: Route) -> ApiError {
395    use bitwarden_api_base::Error;
396
397    match err {
398        Error::Response(rc) => {
399            let status = rc.status.as_u16();
400            match status {
401                401 => {
402                    // No async context to read session.phase(), so a post-retry 401 maps to
403                    // Transient; terminal loss surfaces earlier.
404                    ApiError::Transient("HTTP 401 (post-retry)".to_string())
405                }
406                404 => match route {
407                    Route::AccessConnectorOrJob => ApiError::NotEligible,
408                    Route::Attempt => ApiError::UnknownAttempt,
409                },
410                409 => ApiError::Rejected { status },
411                429 | 500..=599 => ApiError::Transient(format!("HTTP {status}")),
412                other => ApiError::Transient(format!("HTTP {other}")),
413            }
414        }
415        Error::ReqwestMiddleware(mw_err) => {
416            // SessionError stays typed inside the middleware error.
417            if let reqwest_middleware::Error::Middleware(inner) = &mw_err
418                && let Some(SessionError::Lost(l)) = inner.downcast_ref::<SessionError>()
419            {
420                ApiError::SessionLost(*l)
421            } else {
422                ApiError::Transient(format!(
423                    "middleware: {}",
424                    safe_middleware_description(&mw_err)
425                ))
426            }
427        }
428        Error::Reqwest(_) => ApiError::Transient("transport error".to_owned()),
429        Error::Serde(_) => ApiError::Protocol("response decode failed".to_owned()),
430        Error::Io(_) => ApiError::Transient("I/O error".to_owned()),
431    }
432}
433
434/// Names only the variant of a [`reqwest_middleware::Error`]; its message can wrap arbitrary
435/// strings.
436fn safe_middleware_description(err: &reqwest_middleware::Error) -> &'static str {
437    match err {
438        reqwest_middleware::Error::Middleware(_) => "middleware error",
439        reqwest_middleware::Error::Reqwest(_) => "reqwest error",
440    }
441}
442
443/// Parse the cipher response into a vault [`Cipher`]. A conversion failure is a protocol error that
444/// never includes the payload.
445fn parse_cipher(model: CipherMiniResponseModel) -> Result<Cipher, ApiError> {
446    Cipher::try_from(model)
447        .map_err(|_| ApiError::Protocol("cipher response could not be parsed".to_owned()))
448}
449
450/// The connector never constructs an `Unknown` variant, so this guard failure is a bug rather than
451/// a server response.
452fn unrecognized_variant(_: bitwarden_pam::RotationError) -> ApiError {
453    ApiError::Protocol("unrecognized variant".to_owned())
454}
455
456#[cfg(test)]
457mod tests {
458    use std::sync::Arc;
459
460    use bitwarden_crypto::KeyStore;
461    use reqwest_middleware::ClientBuilder;
462    use tokio::{sync::watch, time::Instant};
463    use wiremock::{
464        Mock, MockServer, ResponseTemplate,
465        matchers::{header, method, path},
466    };
467
468    use super::*;
469    use crate::{
470        auth::{
471            identity::IdentityClient,
472            session::{SessionLost, SessionManager},
473        },
474        error::SafeDetail,
475        test_support::{access_token_tagged, encrypted_payload_for, test_org_id, test_token},
476    };
477
478    /// `bearer` is a short tag, expanded to a real JWT so the `organization` claim parses.
479    fn identity_success_response(
480        bearer: &str,
481        expires_in: u64,
482        encrypted_payload: &str,
483    ) -> ResponseTemplate {
484        let bearer = access_token_tagged(test_org_id(), bearer);
485        let body = format!(
486            r#"{{"access_token":"{bearer}","expires_in":{expires_in},"encrypted_payload":"{encrypted_payload}"}}"#
487        );
488        ResponseTemplate::new(200)
489            .set_body_string(body)
490            .insert_header("content-type", "application/json")
491    }
492
493    async fn make_session(identity_server: &MockServer, bearer: &str) -> Arc<SessionManager> {
494        let org_key = bitwarden_crypto::SymmetricCryptoKey::make(
495            bitwarden_crypto::SymmetricKeyAlgorithm::Aes256CbcHmac,
496        );
497        let payload = encrypted_payload_for(&org_key);
498
499        Mock::given(method("POST"))
500            .and(path("/connect/token"))
501            .respond_with(identity_success_response(bearer, 3600, &payload))
502            .mount(identity_server)
503            .await;
504
505        let identity_client = IdentityClient::new(identity_server.uri()).expect("identity client");
506        SessionManager::new(identity_client, test_token(), KeyStore::default())
507            .await
508            .expect("SessionManager::new")
509    }
510
511    /// Like [`make_session`], but only the first token request succeeds; later requests fall
512    /// through to whatever the test mounts next.
513    async fn make_session_once(identity_server: &MockServer, bearer: &str) -> Arc<SessionManager> {
514        let org_key = bitwarden_crypto::SymmetricCryptoKey::make(
515            bitwarden_crypto::SymmetricKeyAlgorithm::Aes256CbcHmac,
516        );
517        let payload = encrypted_payload_for(&org_key);
518
519        Mock::given(method("POST"))
520            .and(path("/connect/token"))
521            .respond_with(identity_success_response(bearer, 3600, &payload))
522            .up_to_n_times(1)
523            .mount(identity_server)
524            .await;
525
526        let identity_client = IdentityClient::new(identity_server.uri()).expect("identity client");
527        SessionManager::new(identity_client, test_token(), KeyStore::default())
528            .await
529            .expect("SessionManager::new")
530    }
531
532    /// A handler holding `session`, as startup leaves it.
533    fn handler_for(session: Arc<SessionManager>) -> AccessConnectorTokenHandler {
534        let handler = AccessConnectorTokenHandler::default();
535        handler.set_session(session);
536        handler
537    }
538
539    /// An [`AccessConnectorClient`] over the production client stack with `session` installed, plus
540    /// the connectivity receiver.
541    fn make_access_connector_client(
542        api_server: &MockServer,
543        session: Arc<SessionManager>,
544    ) -> (AccessConnectorClient, watch::Receiver<Instant>) {
545        let (tx, rx) = watch::channel(Instant::now());
546        let (client, handler) = build_sdk_client(
547            api_server.uri(),
548            "http://identity.invalid".to_owned(),
549            REQUEST_TIMEOUT,
550            tx,
551        );
552        handler.set_session(session);
553        (client.access_connector(), rx)
554    }
555
556    #[test]
557    fn claim_409_maps_to_ok_none() {
558        // claim() intercepts a 409 before classify_error, which on its own maps it to Rejected.
559        let rc = bitwarden_api_base::ResponseContent {
560            status: reqwest::StatusCode::CONFLICT,
561            message: String::new(),
562        };
563        let err: bitwarden_api_base::Error = bitwarden_api_base::Error::Response(rc);
564        let api_err = classify_error(err, Route::AccessConnectorOrJob);
565        assert!(matches!(api_err, ApiError::Rejected { status: 409 }));
566    }
567
568    #[test]
569    fn parse_cipher_parses_a_login() {
570        let json = crate::test_support::keyed_login_json();
571        let model: CipherMiniResponseModel = serde_json::from_value(json).expect("model");
572
573        let cipher = parse_cipher(model).expect("parse_cipher");
574        assert_eq!(
575            Uuid::from(cipher.id.expect("id")),
576            crate::test_support::KEYED_LOGIN_ID
577        );
578        assert_eq!(cipher.r#type, bitwarden_vault::CipherType::Login);
579        assert!(cipher.key.is_some());
580        assert!(cipher.login.is_some());
581    }
582
583    #[test]
584    fn parse_cipher_missing_type_is_protocol_error() {
585        let mut json = crate::test_support::keyed_login_json();
586        json.as_object_mut().unwrap().remove("type");
587        let model: CipherMiniResponseModel = serde_json::from_value(json).expect("model");
588
589        let err = parse_cipher(model).expect_err("should fail");
590        assert!(matches!(err, ApiError::Protocol(_)), "got: {err:?}");
591    }
592
593    /// The vault parses an unparseable field leniently; the strict decrypt in the rotation write
594    /// rejects it (see `bitwarden_pam::rotation::cipher`'s tests), so parsing must not drop or
595    /// echo the value.
596    #[test]
597    fn parse_cipher_keeps_garbage_enc_string_for_strict_decryption() {
598        let store = crate::test_support::org_store();
599        let mut json = crate::test_support::keyed_login_json();
600        json["name"] = "2.secret!|garbage!|payload!".into();
601        let model: CipherMiniResponseModel = serde_json::from_value(json).expect("model");
602
603        let cipher = parse_cipher(model).expect("lenient parse");
604        assert!(cipher.name.is_some());
605        let org_id = crate::test_support::test_org_id();
606        assert!(bitwarden_pam::rotate_cipher_password(&store, org_id, cipher, "new-pw").is_err());
607    }
608
609    #[tokio::test]
610    async fn poll_happy_path_bumps_connectivity_and_parses_jobs() {
611        let identity_server = MockServer::start().await;
612        let api_server = MockServer::start().await;
613
614        let session = make_session(&identity_server, "test-bearer").await;
615        let (api, mut connectivity_rx) = make_access_connector_client(&api_server, session);
616
617        let before = *connectivity_rx.borrow();
618
619        let job_id = Uuid::new_v4();
620        Mock::given(method("GET"))
621            .and(path("/access-connectors/rotation/jobs"))
622            .respond_with(
623                ResponseTemplate::new(200)
624                    .set_body_json(serde_json::json!({
625                        "data": [{"jobId": job_id.to_string(), "targetSystemId": Uuid::new_v4().to_string()}]
626                    }))
627                    .insert_header("content-type", "application/json"),
628            )
629            .mount(&api_server)
630            .await;
631
632        let jobs = api.poll_jobs().await.expect("poll_jobs");
633        assert_eq!(jobs.len(), 1);
634        assert_eq!(jobs[0].id, job_id);
635
636        connectivity_rx
637            .changed()
638            .await
639            .expect("connectivity changed");
640        let after = *connectivity_rx.borrow();
641        assert!(
642            after > before,
643            "connectivity should be bumped after successful poll"
644        );
645    }
646
647    #[tokio::test]
648    async fn poll_404_maps_to_not_eligible() {
649        let identity_server = MockServer::start().await;
650        let api_server = MockServer::start().await;
651
652        let session = make_session(&identity_server, "test-bearer").await;
653        let (api, _rx) = make_access_connector_client(&api_server, session);
654
655        Mock::given(method("GET"))
656            .and(path("/access-connectors/rotation/jobs"))
657            .respond_with(ResponseTemplate::new(404))
658            .mount(&api_server)
659            .await;
660
661        let err = api.poll_jobs().await.expect_err("should fail");
662        assert!(matches!(err, ApiError::NotEligible), "got: {err:?}");
663    }
664
665    #[tokio::test]
666    async fn claim_200_parses_work_snapshot() {
667        let identity_server = MockServer::start().await;
668        let api_server = MockServer::start().await;
669
670        let session = make_session(&identity_server, "test-bearer").await;
671        let (api, _rx) = make_access_connector_client(&api_server, session);
672
673        let job_id = Uuid::new_v4();
674        let attempt_id = Uuid::new_v4();
675        let target_system_id = Uuid::new_v4();
676        let cipher_id = Uuid::new_v4();
677        let execute_by = chrono::Utc::now()
678            .checked_add_signed(chrono::Duration::minutes(5))
679            .unwrap()
680            .to_rfc3339();
681
682        Mock::given(method("POST"))
683            .and(path(format!(
684                "/access-connectors/rotation/jobs/{job_id}/claim"
685            )))
686            .respond_with(
687                ResponseTemplate::new(200)
688                    .set_body_json(serde_json::json!({
689                        "attemptId": attempt_id,
690                        "jobId": job_id,
691                        "targetSystemId": target_system_id,
692                        "targetSystemName": "my-script",
693                        "targetSystemSealedData": "opaque-target-blob",
694                        "targetSystemWrappedKey": "opaque-target-key",
695                        "cipherId": cipher_id,
696                        "rotationConfigSealedData": "opaque-config-blob",
697                        "rotationConfigWrappedKey": "opaque-config-key",
698                        "executeBy": execute_by
699                    }))
700                    .insert_header("content-type", "application/json"),
701            )
702            .mount(&api_server)
703            .await;
704
705        let snap = api
706            .claim(job_id)
707            .await
708            .expect("claim")
709            .expect("some snapshot");
710        assert_eq!(snap.attempt_id, attempt_id);
711        assert_eq!(
712            snap.target_system_sealed_data.as_deref(),
713            Some("opaque-target-blob")
714        );
715        assert_eq!(
716            snap.target_system_wrapped_key.as_deref(),
717            Some("opaque-target-key")
718        );
719        assert_eq!(
720            snap.rotation_config_sealed_data.as_deref(),
721            Some("opaque-config-blob")
722        );
723        assert_eq!(
724            snap.rotation_config_wrapped_key.as_deref(),
725            Some("opaque-config-key")
726        );
727    }
728
729    #[tokio::test]
730    async fn claim_409_maps_to_ok_none_integrated() {
731        let identity_server = MockServer::start().await;
732        let api_server = MockServer::start().await;
733
734        let session = make_session(&identity_server, "test-bearer").await;
735        let (api, _rx) = make_access_connector_client(&api_server, session);
736
737        let job_id = Uuid::new_v4();
738        Mock::given(method("POST"))
739            .and(path(format!(
740                "/access-connectors/rotation/jobs/{job_id}/claim"
741            )))
742            .respond_with(ResponseTemplate::new(409))
743            .mount(&api_server)
744            .await;
745
746        let result = api.claim(job_id).await.expect("no error");
747        assert!(result.is_none(), "409 should map to Ok(None)");
748    }
749
750    #[tokio::test]
751    async fn get_cipher_parses_the_cipher_response() {
752        let identity_server = MockServer::start().await;
753        let api_server = MockServer::start().await;
754
755        let session = make_session(&identity_server, "test-bearer").await;
756        let (api, _rx) = make_access_connector_client(&api_server, session);
757
758        let attempt_id = Uuid::new_v4();
759        let json = crate::test_support::keyed_login_json();
760
761        Mock::given(method("GET"))
762            .and(path(format!(
763                "/access-connectors/rotation/attempts/{attempt_id}/cipher"
764            )))
765            .respond_with(
766                ResponseTemplate::new(200)
767                    .set_body_json(json)
768                    .insert_header("content-type", "application/json"),
769            )
770            .mount(&api_server)
771            .await;
772
773        let cipher = api.get_cipher(attempt_id).await.expect("get_cipher");
774        assert_eq!(
775            Uuid::from(cipher.id.expect("id")),
776            crate::test_support::KEYED_LOGIN_ID
777        );
778        assert_eq!(
779            cipher.revision_date,
780            crate::test_support::CIPHER_REVISION_DATE
781                .parse::<chrono::DateTime<chrono::Utc>>()
782                .unwrap()
783        );
784    }
785
786    #[tokio::test]
787    async fn put_cipher_409_maps_to_rejected() {
788        let identity_server = MockServer::start().await;
789        let api_server = MockServer::start().await;
790
791        let session = make_session(&identity_server, "test-bearer").await;
792        let (api, _rx) = make_access_connector_client(&api_server, session);
793
794        let attempt_id = Uuid::new_v4();
795        Mock::given(method("PUT"))
796            .and(path(format!(
797                "/access-connectors/rotation/attempts/{attempt_id}/cipher"
798            )))
799            .respond_with(ResponseTemplate::new(409))
800            .mount(&api_server)
801            .await;
802
803        let err = api
804            .put_cipher(attempt_id, CipherRequestModel::default())
805            .await
806            .expect_err("should fail");
807
808        assert!(
809            matches!(err, ApiError::Rejected { status: 409 }),
810            "got: {err:?}"
811        );
812    }
813
814    #[tokio::test]
815    async fn put_cipher_404_maps_to_unknown_attempt() {
816        let identity_server = MockServer::start().await;
817        let api_server = MockServer::start().await;
818
819        let session = make_session(&identity_server, "test-bearer").await;
820        let (api, _rx) = make_access_connector_client(&api_server, session);
821
822        let attempt_id = Uuid::new_v4();
823        Mock::given(method("PUT"))
824            .and(path(format!(
825                "/access-connectors/rotation/attempts/{attempt_id}/cipher"
826            )))
827            .respond_with(ResponseTemplate::new(404))
828            .mount(&api_server)
829            .await;
830
831        let err = api
832            .put_cipher(attempt_id, CipherRequestModel::default())
833            .await
834            .expect_err("should fail");
835
836        assert!(matches!(err, ApiError::UnknownAttempt), "got: {err:?}");
837    }
838
839    #[tokio::test]
840    async fn failure_report_serialises_integer_sync_state_and_snake_case_error_code() {
841        let identity_server = MockServer::start().await;
842        let api_server = MockServer::start().await;
843
844        let session = make_session(&identity_server, "test-bearer").await;
845        let (api, _rx) = make_access_connector_client(&api_server, session);
846
847        let attempt_id = Uuid::new_v4();
848
849        Mock::given(method("POST"))
850            .and(path(format!(
851                "/access-connectors/rotation/attempts/{attempt_id}/failure"
852            )))
853            .respond_with(ResponseTemplate::new(200))
854            .mount(&api_server)
855            .await;
856
857        api.report_failure(
858            attempt_id,
859            RotationFailureCode::TargetUnreachable,
860            Some(SafeDetail::from_status(503)),
861            RotationSyncState::TargetUpdated,
862        )
863        .await
864        .expect("report_failure");
865
866        let requests = api_server.received_requests().await.unwrap();
867        assert_eq!(requests.len(), 1);
868        let body: serde_json::Value =
869            serde_json::from_slice(&requests[0].body).expect("parse request body");
870
871        // 1 = TargetUpdated.
872        assert_eq!(
873            body["syncState"],
874            serde_json::Value::Number(serde_json::Number::from(1)),
875            "syncState must be an integer, got: {}",
876            body["syncState"]
877        );
878
879        assert_eq!(
880            body["errorCode"],
881            serde_json::Value::String("target_unreachable".to_owned()),
882            "errorCode must be snake_case string"
883        );
884    }
885
886    #[tokio::test]
887    async fn no_auth_header_without_auth_required_extension() {
888        let identity_server = MockServer::start().await;
889        let api_server = MockServer::start().await;
890
891        let session = make_session(&identity_server, "my-bearer").await;
892        let (_tx, _rx) = watch::channel(Instant::now());
893
894        // A raw client, so requests carry no AuthRequired extension.
895        let http_client = bitwarden_api_base::new_http_client_builder()
896            .timeout(Duration::from_secs(30))
897            .build()
898            .unwrap();
899        let client = ClientBuilder::new(http_client)
900            .with(AccessConnectorAuthMiddleware::new(handler_for(Arc::clone(
901                &session,
902            ))))
903            .build();
904
905        Mock::given(method("GET"))
906            .and(path("/test"))
907            .respond_with(ResponseTemplate::new(200))
908            .mount(&api_server)
909            .await;
910
911        client
912            .get(format!("{}/test", api_server.uri()))
913            .send()
914            .await
915            .expect("request");
916
917        let requests = api_server.received_requests().await.unwrap();
918        assert_eq!(requests.len(), 1);
919        assert!(
920            requests[0].headers.get("authorization").is_none(),
921            "no Authorization header should be attached when AuthRequired extension is absent"
922        );
923    }
924
925    #[tokio::test]
926    async fn middleware_refreshes_once_on_401_and_retries() {
927        use bitwarden_crypto::{SymmetricCryptoKey, SymmetricKeyAlgorithm};
928
929        let identity_server = MockServer::start().await;
930        let api_server = MockServer::start().await;
931        let org_key1 = SymmetricCryptoKey::make(SymmetricKeyAlgorithm::Aes256CbcHmac);
932        let payload1 = encrypted_payload_for(&org_key1);
933        let org_key2 = SymmetricCryptoKey::make(SymmetricKeyAlgorithm::Aes256CbcHmac);
934        let payload2 = encrypted_payload_for(&org_key2);
935
936        Mock::given(method("POST"))
937            .and(path("/connect/token"))
938            .respond_with(identity_success_response("bearer-1", 3600, &payload1))
939            .up_to_n_times(1)
940            .mount(&identity_server)
941            .await;
942        Mock::given(method("POST"))
943            .and(path("/connect/token"))
944            .respond_with(identity_success_response("bearer-2", 3600, &payload2))
945            .mount(&identity_server)
946            .await;
947
948        let identity_client = IdentityClient::new(identity_server.uri()).expect("identity client");
949        let session = SessionManager::new(identity_client, test_token(), KeyStore::default())
950            .await
951            .expect("SessionManager::new");
952
953        let bearer_1 = format!("Bearer {}", access_token_tagged(test_org_id(), "bearer-1"));
954        let bearer_2 = format!("Bearer {}", access_token_tagged(test_org_id(), "bearer-2"));
955
956        Mock::given(method("GET"))
957            .and(path("/access-connectors/rotation/jobs"))
958            .and(header("Authorization", bearer_1.as_str()))
959            .respond_with(ResponseTemplate::new(401))
960            .up_to_n_times(1)
961            .mount(&api_server)
962            .await;
963        Mock::given(method("GET"))
964            .and(path("/access-connectors/rotation/jobs"))
965            .and(header("Authorization", bearer_2.as_str()))
966            .respond_with(
967                ResponseTemplate::new(200)
968                    .set_body_json(serde_json::json!({"data": []}))
969                    .insert_header("content-type", "application/json"),
970            )
971            .mount(&api_server)
972            .await;
973
974        let (api, _rx) = make_access_connector_client(&api_server, session);
975
976        let jobs = api.poll_jobs().await.expect("poll_jobs after 401-refresh");
977        assert!(jobs.is_empty());
978
979        let identity_reqs = identity_server.received_requests().await.unwrap();
980        assert_eq!(
981            identity_reqs.len(),
982            2,
983            "exactly 2 identity calls expected (initial auth + one refresh)"
984        );
985        let api_reqs = api_server.received_requests().await.unwrap();
986        assert_eq!(
987            api_reqs.len(),
988            2,
989            "exactly 2 API requests expected (401 attempt + retry)"
990        );
991    }
992
993    /// Mounts a 401 on the jobs endpoint so every call forces a refresh.
994    async fn mount_jobs_401(api_server: &MockServer) {
995        Mock::given(method("GET"))
996            .and(path("/access-connectors/rotation/jobs"))
997            .respond_with(ResponseTemplate::new(401))
998            .mount(api_server)
999            .await;
1000    }
1001
1002    /// Issue a request through the HTTP stack [`build_sdk_client`] assembled.
1003    async fn get_through_client(
1004        client: &Client,
1005        url: String,
1006    ) -> reqwest_middleware::Result<reqwest::Response> {
1007        client
1008            .internal
1009            .get_api_configurations()
1010            .api_config
1011            .client
1012            .get(url)
1013            .send()
1014            .await
1015    }
1016
1017    /// Sends a GET to `url` through a fresh client and reports whether connectivity was bumped.
1018    async fn bumps_connectivity(url: String) -> bool {
1019        let (tx, mut rx) = watch::channel(Instant::now());
1020        let (client, _handler) = build_sdk_client(
1021            "http://api.invalid".to_owned(),
1022            "http://identity.invalid".to_owned(),
1023            Duration::from_secs(5),
1024            tx,
1025        );
1026        let before = *rx.borrow_and_update();
1027
1028        let _ = get_through_client(&client, url).await;
1029
1030        rx.has_changed().expect("sender alive") && *rx.borrow() > before
1031    }
1032
1033    #[tokio::test]
1034    async fn connectivity_is_bumped_by_a_2xx_even_with_an_unparseable_body() {
1035        let server = MockServer::start().await;
1036        Mock::given(method("GET"))
1037            .and(path("/ok"))
1038            .respond_with(ResponseTemplate::new(200).set_body_string("not json"))
1039            .mount(&server)
1040            .await;
1041
1042        assert!(bumps_connectivity(format!("{}/ok", server.uri())).await);
1043    }
1044
1045    #[tokio::test]
1046    async fn connectivity_is_not_bumped_by_non_2xx_responses() {
1047        let server = MockServer::start().await;
1048        for status in [302, 401, 404, 409, 500, 503] {
1049            Mock::given(method("GET"))
1050                .and(path(format!("/status/{status}")))
1051                .respond_with(ResponseTemplate::new(status))
1052                .mount(&server)
1053                .await;
1054
1055            assert!(
1056                !bumps_connectivity(format!("{}/status/{status}", server.uri())).await,
1057                "{status} must not count as server contact"
1058            );
1059        }
1060    }
1061
1062    #[tokio::test]
1063    async fn connectivity_is_not_bumped_by_a_transport_error() {
1064        // Nothing listens on port 1.
1065        assert!(!bumps_connectivity("http://127.0.0.1:1/x".to_owned()).await);
1066    }
1067
1068    #[tokio::test]
1069    async fn connectivity_is_not_bumped_by_a_timeout() {
1070        let server = MockServer::start().await;
1071        Mock::given(method("GET"))
1072            .and(path("/stall"))
1073            .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(30)))
1074            .mount(&server)
1075            .await;
1076        let (tx, rx) = watch::channel(Instant::now());
1077        let (client, _handler) = build_sdk_client(
1078            server.uri(),
1079            "http://identity.invalid".to_owned(),
1080            Duration::from_millis(100),
1081            tx,
1082        );
1083        let before = *rx.borrow();
1084
1085        get_through_client(&client, format!("{}/stall", server.uri()))
1086            .await
1087            .expect_err("times out");
1088
1089        assert_eq!(*rx.borrow(), before);
1090    }
1091
1092    #[tokio::test]
1093    async fn the_client_exposes_access_connector_sharing_its_key_store() {
1094        use bitwarden_core::key_management::SymmetricKeySlotId;
1095
1096        let (tx, _rx) = watch::channel(Instant::now());
1097        let (client, handler) = build_sdk_client(
1098            "http://api.invalid".to_owned(),
1099            "http://identity.invalid".to_owned(),
1100            REQUEST_TIMEOUT,
1101            tx,
1102        );
1103
1104        // The cipher is encrypted under the vectors' org key; only the handler's store gets the
1105        // key.
1106        let json = crate::test_support::keyed_login_json();
1107        let model: CipherMiniResponseModel = serde_json::from_value(json).expect("model");
1108        let cipher = parse_cipher(model).expect("parse");
1109        let org_id = crate::test_support::test_org_id();
1110
1111        let access_connector = client.access_connector();
1112        assert!(
1113            bitwarden_pam::rotate_cipher_password(
1114                &access_connector.key_store,
1115                org_id,
1116                cipher.clone(),
1117                "new-pw"
1118            )
1119            .is_err(),
1120            "no key yet"
1121        );
1122
1123        let org_key = crate::test_support::fixture_org_key();
1124        let store = handler.key_store().expect("key store");
1125        #[allow(deprecated)]
1126        store
1127            .context_mut()
1128            .set_symmetric_key(SymmetricKeySlotId::Organization(org_id), org_key)
1129            .expect("set key");
1130
1131        // The key reached a client handed out earlier and one handed out later.
1132        for c in [access_connector, client.access_connector()] {
1133            bitwarden_pam::rotate_cipher_password(&c.key_store, org_id, cipher.clone(), "new-pw")
1134                .expect("rotates with the shared store");
1135        }
1136    }
1137
1138    /// A black-holed connection must time out.
1139    #[tokio::test]
1140    async fn request_times_out_rather_than_hanging() {
1141        let api_server = MockServer::start().await;
1142        Mock::given(method("GET"))
1143            .and(path("/stall"))
1144            .respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(30)))
1145            .mount(&api_server)
1146            .await;
1147
1148        let (client, _handler) = build_sdk_client(
1149            api_server.uri(),
1150            "http://identity.invalid".to_string(),
1151            Duration::from_millis(100),
1152            watch::channel(Instant::now()).0,
1153        );
1154
1155        let started = Instant::now();
1156        let err = get_through_client(&client, format!("{}/stall", api_server.uri()))
1157            .await
1158            .expect_err("a stalled request must fail, not hang");
1159
1160        assert!(
1161            started.elapsed() < Duration::from_secs(5),
1162            "must fail on the timeout, not wait out the response"
1163        );
1164        match err {
1165            reqwest_middleware::Error::Middleware(e) => assert!(
1166                e.downcast_ref::<crate::api::RequestTimedOutError>()
1167                    .is_some(),
1168                "expected RequestTimedOutError, got {e:?}"
1169            ),
1170            other => panic!("expected a middleware error, got {other:?}"),
1171        }
1172    }
1173
1174    /// A cross-host redirect must not be followed.
1175    #[tokio::test]
1176    async fn does_not_follow_a_cross_host_redirect() {
1177        let elsewhere = MockServer::start().await;
1178        Mock::given(method("GET"))
1179            .and(path("/elsewhere"))
1180            .respond_with(ResponseTemplate::new(200))
1181            .mount(&elsewhere)
1182            .await;
1183
1184        let api_server = MockServer::start().await;
1185        Mock::given(method("GET"))
1186            .and(path("/redirect"))
1187            .respond_with(ResponseTemplate::new(302).insert_header(
1188                "location",
1189                format!("{}/elsewhere", elsewhere.uri()).as_str(),
1190            ))
1191            .mount(&api_server)
1192            .await;
1193
1194        let (client, _handler) = build_sdk_client(
1195            api_server.uri(),
1196            "http://identity.invalid".to_string(),
1197            Duration::from_secs(5),
1198            watch::channel(Instant::now()).0,
1199        );
1200
1201        let response = get_through_client(&client, format!("{}/redirect", api_server.uri()))
1202            .await
1203            .expect("the 302 itself is a successful exchange");
1204
1205        assert_eq!(
1206            response.status(),
1207            302,
1208            "the redirect must be returned, not followed"
1209        );
1210        assert!(
1211            elsewhere.received_requests().await.unwrap().is_empty(),
1212            "the redirect target must never be contacted"
1213        );
1214    }
1215
1216    #[tokio::test]
1217    async fn forced_refresh_rejected_maps_to_session_lost_revoked() {
1218        let identity_server = MockServer::start().await;
1219        let api_server = MockServer::start().await;
1220        let session = make_session_once(&identity_server, "bearer-1").await;
1221        // The first token request succeeded; the forced refresh is rejected.
1222        Mock::given(method("POST"))
1223            .and(path("/connect/token"))
1224            .respond_with(
1225                ResponseTemplate::new(400)
1226                    .set_body_string(r#"{"error":"invalid_client"}"#)
1227                    .insert_header("content-type", "application/json"),
1228            )
1229            .mount(&identity_server)
1230            .await;
1231        mount_jobs_401(&api_server).await;
1232
1233        let (api, _rx) = make_access_connector_client(&api_server, session);
1234
1235        let err = api.poll_jobs().await.expect_err("revoked mid-session");
1236        assert!(
1237            matches!(err, ApiError::SessionLost(SessionLost::Revoked)),
1238            "expected SessionLost(Revoked), got {err:?}"
1239        );
1240    }
1241
1242    #[tokio::test]
1243    async fn call_after_session_close_maps_to_session_lost_closed() {
1244        let identity_server = MockServer::start().await;
1245        let api_server = MockServer::start().await;
1246        let session = make_session(&identity_server, "bearer-1").await;
1247        session.close().await;
1248
1249        let (api, _rx) = make_access_connector_client(&api_server, session);
1250
1251        let err = api.poll_jobs().await.expect_err("session closed");
1252        assert!(
1253            matches!(err, ApiError::SessionLost(SessionLost::Closed)),
1254            "expected SessionLost(Closed), got {err:?}"
1255        );
1256        assert!(
1257            api_server.received_requests().await.unwrap().is_empty(),
1258            "no request should reach the API once the session is closed"
1259        );
1260    }
1261
1262    // Runs in real time (~3 s of session backoff): paused time breaks the HTTP client's connect
1263    // timeout.
1264    #[tokio::test]
1265    async fn forced_refresh_transient_failure_maps_to_transient() {
1266        let identity_server = MockServer::start().await;
1267        let api_server = MockServer::start().await;
1268        let session = make_session_once(&identity_server, "bearer-1").await;
1269        // The first token request succeeded; the forced refresh keeps failing with a 5xx.
1270        Mock::given(method("POST"))
1271            .and(path("/connect/token"))
1272            .respond_with(ResponseTemplate::new(503))
1273            .mount(&identity_server)
1274            .await;
1275        mount_jobs_401(&api_server).await;
1276
1277        let (api, _rx) = make_access_connector_client(&api_server, session);
1278
1279        let err = api.poll_jobs().await.expect_err("refresh keeps failing");
1280        assert!(
1281            matches!(err, ApiError::Transient(_)),
1282            "expected Transient, got {err:?}"
1283        );
1284    }
1285}
bitwarden_access_connector/api/mod.rs

bitwarden_access_connector/api/
mod.rs
