pmcp/shared/credential_store.rs
1//! Target-agnostic OAuth credential storage: the key, the record, the document
2//! format, the schema 1 → 2 migration, the platform seam and its administrative
3//! sibling.
4//!
5//! # Why this tier is ungated
6//!
7//! A file under the user's home directory is not a viable credential store for
8//! ANY hosting target — the home directory is unwritable on AWS Lambda and
9//! per-container on Cloudflare Workers and Cloud Run. So credential storage
10//! lands behind a trait, and everything a platform needs in order to implement
11//! that trait lives here: outside the `oauth` feature and outside any target
12//! gate. This module has no `#[cfg]` attribute other than the one over its own
13//! unit tests, performs no I/O of any kind, and imports nothing beyond this
14//! crate's error type, `serde`, `async_trait`, `parking_lot` and `url`.
15//!
16//! The practical consequence: a platform that keeps the same JSON document in
17//! `DynamoDB` or a KV store gets byte-identical parsing, migration and
18//! reporting behaviour to the CLI, because
19//! [`parse_credential_snapshot`](crate::shared::credential_store::parse_credential_snapshot)
20//! and
21//! [`CredentialSnapshot::to_bytes`](crate::shared::credential_store::CredentialSnapshot::to_bytes)
22//! are the ONLY places the on-disk shape is known. A gated file implementation
23//! reduces to lock, read, parse, mutate, serialize, write.
24//!
25//! # The key is three-part
26//!
27//! [`CredentialKey`](crate::shared::credential_store::CredentialKey) is
28//! `(issuer, account, server)`. The issuer component is SEP-2352's requirement
29//! — "clients MUST NOT reuse client credentials from a different authorization
30//! server" holds by construction, with no enforcement branch, because a
31//! different authorization server is simply a different key.
32//!
33//! The `server` component closes a second collision that the two-part form
34//! leaves open: two MCP servers can share one authorization server and one user
35//! account while holding DIFFERENT registrations, different client IDs and
36//! different granted scopes. Under a two-part key they collide — a logout on
37//! one deletes the other's credentials, and a migration can overwrite one with
38//! the other. RFC 8707's `resource` parameter would have bound the audience and
39//! mitigated this; it is deferred by owner decision, so the key carries the
40//! binding instead.
41//!
42//! The `account` component is caller-supplied and NEVER interpreted by this
43//! crate: a Cognito `sub`, a tenant id, or empty for the single-user CLI.
44//!
45//! # Why the public structs have private fields
46//!
47//! `OAuthConfig`, `DcrRequest` and `OidcDiscoveryMetadata` are all-public-field
48//! structs that are not `#[non_exhaustive]`, so adding a field to any of them is
49//! a MAJOR semver event. The types here use private fields with constructors and
50//! accessors precisely so they stay extensible at minor forever — which is what
51//! let `registered_application_type` and `granted_scopes` be added without a
52//! semver event, and what lets the next such field be added the same way. Do not
53//! "simplify" them to public fields.
54//!
55//! # Examples
56//!
57//! ```
58//! use pmcp::{CredentialKey, CredentialSnapshot, StoredCredentials};
59//!
60//! let mut snapshot = CredentialSnapshot::new();
61//! let key = CredentialKey::new("https://as.example", "", "https://mcp.example");
62//! snapshot.insert(key.clone(), StoredCredentials::new("access-token", "client-id"));
63//!
64//! // A different authorization server is a cache MISS, by key shape alone.
65//! let other = CredentialKey::new("https://evil.example", "", "https://mcp.example");
66//! assert!(snapshot.get(&other).is_none());
67//! assert!(snapshot.get(&key).is_some());
68//! # Ok::<(), pmcp::Error>(())
69//! ```
70
71use std::collections::BTreeMap;
72use std::fmt;
73
74use async_trait::async_trait;
75use parking_lot::RwLock;
76use serde::{Deserialize, Serialize};
77use url::Url;
78
79use crate::error::{Error, Result};
80
81/// The document schema version this build reads and writes.
82pub const CREDENTIAL_SCHEMA_VERSION: u32 = 2;
83
84/// The schema version of `cargo-pmcp`'s pre-existing multi-server token cache.
85const LEGACY_SCHEMA_VERSION: u32 = 1;
86
87/// What a redacted secret renders as in [`StoredCredentials`]'s `Debug`.
88const REDACTED: &str = "<redacted>";
89
90/// Why an entry could not be re-keyed during a schema 1 → 2 migration.
91const MISSING_ISSUER_REASON: &str =
92 "entry records no issuer; it cannot be re-keyed without guessing which \
93 authorization server issued it, so it was dropped rather than misattributed";
94
95/// account → credentials, for one issuer and one server.
96type AccountMap = BTreeMap<String, StoredCredentials>;
97/// server → [`AccountMap`], for one issuer.
98type ServerMap = BTreeMap<String, AccountMap>;
99/// issuer → [`ServerMap`] — the whole credential tree.
100type IssuerMap = BTreeMap<String, ServerMap>;
101
102// ---------------------------------------------------------------------------
103// Key
104// ---------------------------------------------------------------------------
105
106/// The address of one stored credential: `(issuer, account, server)`.
107///
108/// All three components are stored verbatim and compared byte-for-byte. No
109/// normalization of any kind is applied here — the `server` component is
110/// expected to already be the value [`normalize_server_key`] produced, so that
111/// trailing-slash and host-case variants of one MCP server URL do not become
112/// two keys.
113///
114/// # Why three components and not two
115///
116/// - **issuer** — SEP-2352: credentials obtained from one authorization server
117/// MUST NOT be reused with another. Including the issuer makes that true by
118/// construction rather than by an enforcement branch somebody can forget.
119/// - **server** — two MCP servers can share one authorization server AND one
120/// account while holding different registrations, different client IDs and
121/// different granted scopes. Without this component they collide: a logout on
122/// one deletes the other's credentials, and a schema migration can overwrite
123/// one with the other. RFC 8707's `resource` parameter would have bound the
124/// audience and mitigated this; it is deferred, so the key carries it.
125/// - **account** — caller-supplied and never interpreted: a Cognito `sub`, a
126/// tenant id, or the empty string for the single-user CLI.
127///
128/// # Examples
129///
130/// ```
131/// use pmcp::CredentialKey;
132///
133/// let key = CredentialKey::new("https://as.example", "sub-abc|123", "https://mcp.example");
134/// assert_eq!(key.issuer(), "https://as.example");
135/// assert_eq!(key.account(), "sub-abc|123");
136/// assert_eq!(key.server(), "https://mcp.example");
137///
138/// // Two MCP servers sharing one authorization server and one account do NOT collide.
139/// let other = CredentialKey::new("https://as.example", "sub-abc|123", "https://other.example");
140/// assert_ne!(key, other);
141/// ```
142#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
143pub struct CredentialKey {
144 issuer: String,
145 account: String,
146 server: String,
147}
148
149impl CredentialKey {
150 /// Build a key from its three components. None of them is validated,
151 /// normalized or interpreted.
152 pub fn new<I, A, S>(issuer: I, account: A, server: S) -> Self
153 where
154 I: Into<String>,
155 A: Into<String>,
156 S: Into<String>,
157 {
158 Self {
159 issuer: issuer.into(),
160 account: account.into(),
161 server: server.into(),
162 }
163 }
164
165 /// The authorization server's `issuer` identifier.
166 pub fn issuer(&self) -> &str {
167 &self.issuer
168 }
169
170 /// The caller-supplied account scope. Empty for the single-user CLI.
171 pub fn account(&self) -> &str {
172 &self.account
173 }
174
175 /// The normalized MCP server key. See [`normalize_server_key`].
176 pub fn server(&self) -> &str {
177 &self.server
178 }
179}
180
181// ---------------------------------------------------------------------------
182// Record
183// ---------------------------------------------------------------------------
184
185/// The credentials held for one [`CredentialKey`].
186///
187/// Field names on the wire are the `snake_case` names `cargo-pmcp`'s existing
188/// multi-server cache already uses — `access_token`, `refresh_token`,
189/// `expires_at`, `scopes`, `client_id` — plus the new optional
190/// `registered_application_type`. That is not cosmetic: the pre-existing file is
191/// the migration source AND the surviving path, so a field-name divergence here
192/// would silently drop data.
193///
194/// `Debug` is implemented BY HAND. A derived one would put both bearer tokens
195/// into every log line and panic message that formats a record.
196///
197/// # Examples
198///
199/// ```
200/// use pmcp::StoredCredentials;
201///
202/// let record = StoredCredentials::new("access-token", "client-id")
203/// .with_refresh_token("refresh-token")
204/// .with_granted_scopes(["mcp:read", "mcp:write"]);
205///
206/// assert_eq!(record.granted_scopes(), ["mcp:read", "mcp:write"]);
207/// // Neither token survives into the Debug rendering.
208/// assert!(!format!("{record:?}").contains("access-token"));
209/// ```
210#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
211pub struct StoredCredentials {
212 access_token: String,
213 #[serde(default, skip_serializing_if = "Option::is_none")]
214 refresh_token: Option<String>,
215 #[serde(default, skip_serializing_if = "Option::is_none")]
216 expires_at: Option<u64>,
217 #[serde(default, skip_serializing_if = "Vec::is_empty")]
218 scopes: Vec<String>,
219 client_id: String,
220 #[serde(default, skip_serializing_if = "Option::is_none")]
221 registered_application_type: Option<String>,
222}
223
224impl StoredCredentials {
225 /// Build a record from the two values every successful flow produces.
226 pub fn new(access_token: impl Into<String>, client_id: impl Into<String>) -> Self {
227 Self {
228 access_token: access_token.into(),
229 refresh_token: None,
230 expires_at: None,
231 scopes: Vec::new(),
232 client_id: client_id.into(),
233 registered_application_type: None,
234 }
235 }
236
237 /// Attach the refresh token, when the authorization server issued one.
238 pub fn with_refresh_token(mut self, refresh_token: impl Into<String>) -> Self {
239 self.refresh_token = Some(refresh_token.into());
240 self
241 }
242
243 /// Attach the absolute expiry, in Unix seconds.
244 pub fn with_expires_at(mut self, expires_at: u64) -> Self {
245 self.expires_at = Some(expires_at);
246 self
247 }
248
249 /// Attach the GRANTED scopes from the token response, in order.
250 pub fn with_granted_scopes<S, I>(mut self, scopes: I) -> Self
251 where
252 I: IntoIterator<Item = S>,
253 S: Into<String>,
254 {
255 self.scopes = scopes.into_iter().map(Into::into).collect();
256 self
257 }
258
259 /// Attach the `application_type` the authorization server actually
260 /// registered, which may differ from the one that was requested.
261 pub fn with_registered_application_type(mut self, application_type: impl Into<String>) -> Self {
262 self.registered_application_type = Some(application_type.into());
263 self
264 }
265
266 /// The bearer access token. Sensitive — never log it.
267 pub fn access_token(&self) -> &str {
268 &self.access_token
269 }
270
271 /// The refresh token, when one was issued. Sensitive — never log it.
272 pub fn refresh_token(&self) -> Option<&str> {
273 self.refresh_token.as_deref()
274 }
275
276 /// Absolute expiry in Unix seconds, when the response carried one.
277 pub fn expires_at(&self) -> Option<u64> {
278 self.expires_at
279 }
280
281 /// The GRANTED scopes, verbatim and in the order the response listed them.
282 pub fn granted_scopes(&self) -> &[String] {
283 &self.scopes
284 }
285
286 /// The effective client id — issued by dynamic registration, or supplied.
287 pub fn client_id(&self) -> &str {
288 &self.client_id
289 }
290
291 /// The `application_type` the authorization server registered, if observed.
292 pub fn registered_application_type(&self) -> Option<&str> {
293 self.registered_application_type.as_deref()
294 }
295}
296
297impl fmt::Debug for StoredCredentials {
298 /// Redacts both bearer tokens while keeping the shape legible. Presence of
299 /// a refresh token stays observable, because "can this record refresh?" is
300 /// the question a reader is usually asking.
301 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
302 f.debug_struct("StoredCredentials")
303 .field("access_token", &REDACTED)
304 .field(
305 "refresh_token",
306 &self.refresh_token.as_ref().map(|_| REDACTED),
307 )
308 .field("expires_at", &self.expires_at)
309 .field("scopes", &self.scopes)
310 .field("client_id", &self.client_id)
311 .field(
312 "registered_application_type",
313 &self.registered_application_type,
314 )
315 .finish()
316 }
317}
318
319// ---------------------------------------------------------------------------
320// Document
321// ---------------------------------------------------------------------------
322
323/// The whole credential document, in memory.
324///
325/// The credential tree is nested issuer → server → account rather than keyed by
326/// a concatenated composite string, so no separator has to be invented that an
327/// issuer URL could itself contain. Every level is a `BTreeMap`, which is what
328/// makes [`CredentialSnapshot::to_bytes`] byte-stable across writes.
329///
330/// A second, flat map records the last-seen issuer per normalized server key.
331/// Issuer-keyed storage makes an authorization-server substitution SAFE but
332/// invisible; recording the previous issuer is what lets a caller notice and
333/// warn.
334///
335/// # Examples
336///
337/// ```
338/// use pmcp::{CredentialKey, CredentialSnapshot, StoredCredentials};
339///
340/// let mut snapshot = CredentialSnapshot::new();
341/// let key = CredentialKey::new("https://as.example", "", "https://mcp.example");
342/// snapshot.insert(key.clone(), StoredCredentials::new("at", "cid"));
343/// snapshot.record_issuer("https://mcp.example", "https://as.example");
344///
345/// // Serializing the same snapshot twice produces identical bytes.
346/// assert_eq!(snapshot.to_bytes()?, snapshot.to_bytes()?);
347/// assert_eq!(snapshot.keys(), vec![key]);
348/// # Ok::<(), pmcp::Error>(())
349/// ```
350#[derive(Clone, Debug, Default, PartialEq, Eq)]
351pub struct CredentialSnapshot {
352 credentials: IssuerMap,
353 issuers: BTreeMap<String, String>,
354}
355
356impl CredentialSnapshot {
357 /// An empty snapshot.
358 pub fn new() -> Self {
359 Self::default()
360 }
361
362 /// The credentials stored under `key`, if any.
363 pub fn get(&self, key: &CredentialKey) -> Option<&StoredCredentials> {
364 self.credentials
365 .get(&key.issuer)?
366 .get(&key.server)?
367 .get(&key.account)
368 }
369
370 /// Store `credentials` under `key`, replacing anything already there.
371 pub fn insert(&mut self, key: CredentialKey, credentials: StoredCredentials) {
372 self.credentials
373 .entry(key.issuer)
374 .or_default()
375 .entry(key.server)
376 .or_default()
377 .insert(key.account, credentials);
378 }
379
380 /// Remove `key`, returning whether anything was there.
381 ///
382 /// Emptied inner maps are pruned, so [`CredentialSnapshot::keys`] stays
383 /// accurate and the serialized bytes do not accumulate empty objects.
384 pub fn remove(&mut self, key: &CredentialKey) -> bool {
385 let Some(by_server) = self.credentials.get_mut(&key.issuer) else {
386 return false;
387 };
388 let Some(by_account) = by_server.get_mut(&key.server) else {
389 return false;
390 };
391 let removed = by_account.remove(&key.account).is_some();
392 if by_account.is_empty() {
393 by_server.remove(&key.server);
394 }
395 if by_server.is_empty() {
396 self.credentials.remove(&key.issuer);
397 }
398 removed
399 }
400
401 /// Every stored key, in a deterministic order.
402 pub fn keys(&self) -> Vec<CredentialKey> {
403 let mut out = Vec::new();
404 for (issuer, by_server) in &self.credentials {
405 for (server, by_account) in by_server {
406 for account in by_account.keys() {
407 out.push(CredentialKey::new(issuer, account, server));
408 }
409 }
410 }
411 out
412 }
413
414 /// Every stored key whose `server` component equals `server_key`, across
415 /// all issuers and accounts. This is what a per-server logout operates on.
416 pub fn keys_for_server(&self, server_key: &str) -> Vec<CredentialKey> {
417 // The tree is keyed issuer -> server -> account, so the server is a
418 // direct lookup rather than a scan. Building every key in the store and
419 // discarding the non-matching ones cost three String allocations per
420 // stored credential to find the handful under one server.
421 let mut out = Vec::new();
422 for (issuer, by_server) in &self.credentials {
423 if let Some(by_account) = by_server.get(server_key) {
424 for account in by_account.keys() {
425 out.push(CredentialKey::new(issuer, account, server_key));
426 }
427 }
428 }
429 out
430 }
431
432 /// Remove everything, returning how many credentials were removed.
433 ///
434 /// The last-seen-issuer records go too: after a full logout the store must
435 /// not retain a list of which authorization servers the user visited.
436 pub fn clear(&mut self) -> usize {
437 let removed = self.credential_count();
438 self.credentials.clear();
439 self.issuers.clear();
440 removed
441 }
442
443 /// The issuer last seen for `server_key`, if one was ever recorded.
444 pub fn last_issuer(&self, server_key: &str) -> Option<&str> {
445 self.issuers.get(server_key).map(String::as_str)
446 }
447
448 /// Record the issuer currently in use for `server_key`, replacing any
449 /// previous value.
450 pub fn record_issuer(&mut self, server_key: &str, issuer: &str) {
451 self.issuers
452 .insert(server_key.to_owned(), issuer.to_owned());
453 }
454
455 /// Serialize to the current document format.
456 ///
457 /// Byte-stable: serializing the same snapshot twice produces identical
458 /// bytes, because every map is ordered. That is what makes a diff of a
459 /// credential file meaningful and stops an atomic write churning the file
460 /// on every save.
461 pub fn to_bytes(&self) -> Result<Vec<u8>> {
462 let document = DocumentRef {
463 schema_version: CREDENTIAL_SCHEMA_VERSION,
464 credentials: &self.credentials,
465 issuers: &self.issuers,
466 };
467 serde_json::to_vec_pretty(&document)
468 .map_err(|e| Error::internal(format!("failed to serialize credentials: {e}")))
469 }
470
471 /// Forget the last-seen issuer for one server, without touching any other.
472 ///
473 /// `pub(crate)` rather than private so the gated file store in
474 /// `crate::shared::credential_file` gives `delete_by_server` the SAME
475 /// semantics as [`InMemoryCredentialStore`] instead of reimplementing them —
476 /// a per-server logout must not leave behind a record of which
477 /// authorization server the user visited. Deliberately not `pub`: the
478 /// operation only makes sense as part of a delete, and exposing it would
479 /// invite a caller to desynchronize the two maps.
480 pub(crate) fn forget_issuer(&mut self, server_key: &str) {
481 self.issuers.remove(server_key);
482 }
483
484 /// How many credentials are stored, across every issuer and server.
485 fn credential_count(&self) -> usize {
486 self.credentials
487 .values()
488 .map(|by_server| by_server.values().map(BTreeMap::len).sum::<usize>())
489 .sum()
490 }
491}
492
493/// Borrowed serialization view of the current document format.
494#[derive(Serialize)]
495struct DocumentRef<'a> {
496 schema_version: u32,
497 credentials: &'a IssuerMap,
498 issuers: &'a BTreeMap<String, String>,
499}
500
501/// Owned deserialization view of the current document format.
502#[derive(Deserialize)]
503struct Document {
504 #[serde(default)]
505 credentials: IssuerMap,
506 #[serde(default)]
507 issuers: BTreeMap<String, String>,
508}
509
510/// Reads nothing but the version, so dispatch happens before the rest of a
511/// hostile document is interpreted.
512#[derive(Deserialize)]
513struct VersionProbe {
514 schema_version: u32,
515}
516
517/// `cargo-pmcp`'s schema-1 multi-server cache, mirrored for migration only.
518#[derive(Deserialize)]
519struct LegacyCache {
520 #[serde(default)]
521 entries: BTreeMap<String, LegacyEntry>,
522}
523
524/// One schema-1 entry. The map key that addresses it is the normalized MCP
525/// server URL, which is why widening the key is lossless.
526#[derive(Deserialize)]
527struct LegacyEntry {
528 access_token: String,
529 #[serde(default)]
530 refresh_token: Option<String>,
531 #[serde(default)]
532 expires_at: Option<u64>,
533 #[serde(default)]
534 scopes: Vec<String>,
535 #[serde(default)]
536 issuer: Option<String>,
537 #[serde(default)]
538 client_id: String,
539}
540
541impl LegacyEntry {
542 /// Carry every schema-1 field across. `registered_application_type` did not
543 /// exist in schema 1, so it starts absent.
544 fn into_stored(self) -> StoredCredentials {
545 StoredCredentials {
546 access_token: self.access_token,
547 refresh_token: self.refresh_token,
548 expires_at: self.expires_at,
549 scopes: self.scopes,
550 client_id: self.client_id,
551 registered_application_type: None,
552 }
553 }
554}
555
556// ---------------------------------------------------------------------------
557// Migration reporting
558// ---------------------------------------------------------------------------
559
560/// What a parse did, so a caller can tell an operator about it.
561///
562/// Deliberately RETURNED rather than logged: the caller decides between a
563/// `tracing` warning and a line of CLI output.
564#[derive(Clone, Debug, Default, PartialEq, Eq)]
565#[non_exhaustive]
566pub struct MigrationReport {
567 migrated: usize,
568 dropped: Vec<DroppedEntry>,
569}
570
571impl MigrationReport {
572 /// How many entries were re-keyed into the current format.
573 pub fn migrated(&self) -> usize {
574 self.migrated
575 }
576
577 /// Entries that could not be carried across, each naming why.
578 pub fn dropped(&self) -> &[DroppedEntry] {
579 &self.dropped
580 }
581
582 /// Whether the parse changed nothing — the current-version case.
583 pub fn is_noop(&self) -> bool {
584 self.migrated == 0 && self.dropped.is_empty()
585 }
586}
587
588/// One entry a migration could not carry across.
589#[derive(Clone, Debug, PartialEq, Eq)]
590pub struct DroppedEntry {
591 server_key: String,
592 reason: String,
593}
594
595impl DroppedEntry {
596 /// The server key the dropped entry was stored under.
597 pub fn server_key(&self) -> &str {
598 &self.server_key
599 }
600
601 /// Why it could not be carried across, in operator-readable prose.
602 pub fn reason(&self) -> &str {
603 &self.reason
604 }
605}
606
607// ---------------------------------------------------------------------------
608// Parsing
609// ---------------------------------------------------------------------------
610
611/// Parse a credential document, migrating it to the current schema if needed.
612///
613/// This function is TOTAL: it never panics, for any byte sequence. It contains
614/// no indexing and no fallible-unwrapping call, and it is fuzzed over arbitrary
615/// bytes. Refusals name the rule that was violated and never reproduce any byte
616/// of the input, because a credential document is exactly the kind of thing
617/// whose contents must not reach a log.
618///
619/// # Migration
620///
621/// A schema-1 document (`cargo-pmcp`'s multi-server cache) is re-keyed in
622/// memory: each entry that records an issuer becomes
623/// `CredentialKey::new(issuer, "", <the schema-1 map key>)`, and the map key
624/// also becomes that server's last-seen issuer record. An entry that records NO
625/// issuer cannot be re-keyed without guessing which authorization server issued
626/// it — precisely what SEP-2352 forbids — so it is dropped and reported.
627///
628/// # Examples
629///
630/// ```
631/// use pmcp::{parse_credential_snapshot, CredentialKey};
632///
633/// let legacy = br#"{
634/// "schema_version": 1,
635/// "entries": {
636/// "https://mcp.example": {
637/// "access_token": "at",
638/// "issuer": "https://as.example",
639/// "client_id": "cid"
640/// }
641/// }
642/// }"#;
643///
644/// let (snapshot, report) = parse_credential_snapshot(legacy)?;
645/// assert_eq!(report.migrated(), 1);
646/// assert!(report.dropped().is_empty());
647///
648/// let key = CredentialKey::new("https://as.example", "", "https://mcp.example");
649/// assert_eq!(snapshot.get(&key).map(|c| c.client_id()), Some("cid"));
650/// # Ok::<(), pmcp::Error>(())
651/// ```
652pub fn parse_credential_snapshot(bytes: &[u8]) -> Result<(CredentialSnapshot, MigrationReport)> {
653 let probe: VersionProbe = serde_json::from_slice(bytes).map_err(|e| malformed_document(&e))?;
654 match probe.schema_version {
655 CREDENTIAL_SCHEMA_VERSION => parse_current(bytes),
656 LEGACY_SCHEMA_VERSION => migrate_legacy(bytes),
657 observed => Err(unsupported_schema_version(observed)),
658 }
659}
660
661/// Read a document already at the current schema version.
662fn parse_current(bytes: &[u8]) -> Result<(CredentialSnapshot, MigrationReport)> {
663 let document: Document = serde_json::from_slice(bytes).map_err(|e| malformed_document(&e))?;
664 // `Document::credentials` is already an `IssuerMap` — the same type and the
665 // same nesting the snapshot stores — so it is MOVED rather than walked and
666 // re-inserted. Rebuilding it entry-by-entry re-cloned the issuer and server
667 // strings once per account, on a path every load/list/mutation runs through.
668 let snapshot = CredentialSnapshot {
669 credentials: document.credentials,
670 issuers: document.issuers,
671 };
672 Ok((snapshot, MigrationReport::default()))
673}
674
675/// Re-key a schema-1 document into the current format.
676fn migrate_legacy(bytes: &[u8]) -> Result<(CredentialSnapshot, MigrationReport)> {
677 let cache: LegacyCache = serde_json::from_slice(bytes).map_err(|e| malformed_document(&e))?;
678 let mut snapshot = CredentialSnapshot::new();
679 let mut migrated = 0usize;
680 let mut dropped = Vec::new();
681
682 for (server_key, mut entry) in cache.entries {
683 let recorded = entry.issuer.take().filter(|value| !value.is_empty());
684 let Some(issuer) = recorded else {
685 dropped.push(DroppedEntry {
686 server_key,
687 reason: MISSING_ISSUER_REASON.to_owned(),
688 });
689 continue;
690 };
691 snapshot.record_issuer(&server_key, &issuer);
692 snapshot.insert(
693 CredentialKey::new(issuer, "", server_key),
694 entry.into_stored(),
695 );
696 migrated += 1;
697 }
698
699 Ok((snapshot, MigrationReport { migrated, dropped }))
700}
701
702/// A refusal that names the classification and the position, and reproduces no
703/// byte of the input.
704fn malformed_document(err: &serde_json::Error) -> Error {
705 Error::validation(format!(
706 "credential document is malformed: {:?} error at line {}, column {}",
707 err.classify(),
708 err.line(),
709 err.column()
710 ))
711}
712
713/// A refusal that names BOTH the observed and the supported version, and says
714/// what to do about it.
715fn unsupported_schema_version(observed: u32) -> Error {
716 Error::validation(format!(
717 "credential document schema_version {observed} is not supported by this build, \
718 which reads version {CREDENTIAL_SCHEMA_VERSION}; upgrade pmcp to read it"
719 ))
720}
721
722// ---------------------------------------------------------------------------
723// Helper
724// ---------------------------------------------------------------------------
725
726/// Reduce an MCP server URL to one stable key: `scheme://host[:port]`, with the
727/// host lowercased, the path and query dropped, and default ports removed.
728///
729/// This is the value the `server` component of a [`CredentialKey`] is expected
730/// to carry, so that trailing-slash and host-case variants of one MCP server
731/// URL do not become two credentials. It is idempotent.
732///
733/// Refusals do not reproduce the input, which may carry userinfo.
734///
735/// # Examples
736///
737/// ```
738/// use pmcp::shared::credential_store::normalize_server_key;
739///
740/// assert_eq!(normalize_server_key("https://MCP.Example/api/")?, "https://mcp.example");
741/// assert_eq!(normalize_server_key("https://mcp.example:443")?, "https://mcp.example");
742/// assert_eq!(normalize_server_key("https://mcp.example:8443/x")?, "https://mcp.example:8443");
743/// # Ok::<(), pmcp::Error>(())
744/// ```
745pub fn normalize_server_key(server_url: &str) -> Result<String> {
746 let parsed = Url::parse(server_url)
747 .map_err(|e| Error::validation(format!("invalid MCP server URL ({e})")))?;
748 let host = parsed
749 .host_str()
750 .ok_or_else(|| Error::validation("MCP server URL has no host"))?
751 .to_ascii_lowercase();
752
753 let mut key = format!("{}://{}", parsed.scheme(), host);
754 if let Some(port) = parsed.port() {
755 let is_default = (parsed.scheme() == "https" && port == 443)
756 || (parsed.scheme() == "http" && port == 80);
757 if !is_default {
758 key.push_str(&format!(":{port}"));
759 }
760 }
761 Ok(key)
762}
763
764// ---------------------------------------------------------------------------
765// The platform seam
766// ---------------------------------------------------------------------------
767
768/// The narrow seam an OAuth flow needs from credential storage.
769///
770/// This is the trait a HOSTING PLATFORM implements — for `DynamoDB`, a KV
771/// store, a secrets manager — and the one `OAuthHelper` holds. It is
772/// deliberately small: three required methods plus three that default. A store
773/// that vends one user's credentials for one server can implement it in a few
774/// lines and has no business being asked to enumerate or wipe anything; the
775/// administrative operations a command-line tool needs live on
776/// [`CredentialStoreAdmin`] instead.
777///
778/// # What is deliberately NOT here
779///
780/// Token refresh. A trait that owned refresh would need an HTTP client, which
781/// would break both I/O-free construction and this tier's target-cleanliness in
782/// one move. Refresh stays with the OAuth helper, which READS this store for the
783/// client id and the granted scopes it needs.
784///
785/// Construction is likewise I/O-free by contract: an implementor takes every
786/// value it needs as a constructor parameter and reads no environment, no disk
787/// and no network while being built.
788///
789/// # Examples
790///
791/// ```
792/// use pmcp::{CredentialKey, CredentialStore, InMemoryCredentialStore, StoredCredentials};
793///
794/// # async fn demo() -> pmcp::Result<()> {
795/// let store = InMemoryCredentialStore::new();
796/// let key = CredentialKey::new("https://as.example", "", "https://mcp.example");
797///
798/// store.save(&key, &StoredCredentials::new("at", "cid")).await?;
799/// assert!(store.load(&key).await?.is_some());
800///
801/// store.delete(&key).await?;
802/// assert!(store.load(&key).await?.is_none());
803/// # Ok(())
804/// # }
805/// ```
806#[async_trait]
807pub trait CredentialStore: Send + Sync + fmt::Debug {
808 /// Load the credentials stored under `key`, if any.
809 async fn load(&self, key: &CredentialKey) -> Result<Option<StoredCredentials>>;
810
811 /// Store `credentials` under `key`, replacing anything already there.
812 async fn save(&self, key: &CredentialKey, credentials: &StoredCredentials) -> Result<()>;
813
814 /// Remove `key`. Removing a key that is not present is NOT an error.
815 async fn delete(&self, key: &CredentialKey) -> Result<()>;
816
817 /// Save credentials and record the server's issuer in ONE operation.
818 ///
819 /// Doing the two separately leaves a window in which a crash or a lost
820 /// update makes the store claim one issuer while holding another's
821 /// credentials. The DEFAULT implementation here is NOT atomic — it simply
822 /// calls [`CredentialStore::save`] and then
823 /// [`CredentialStore::record_issuer`]. An implementor whose storage can do
824 /// both under one lock or in one transaction SHOULD override it.
825 async fn save_with_issuer(
826 &self,
827 key: &CredentialKey,
828 credentials: &StoredCredentials,
829 server_key: &str,
830 issuer: &str,
831 ) -> Result<()> {
832 self.save(key, credentials).await?;
833 self.record_issuer(server_key, issuer).await
834 }
835
836 /// The issuer last seen for `server_key`.
837 ///
838 /// Defaults to `Ok(None)` so an implementor that does not want last-seen
839 /// issuer tracking is not broken by it. A store that returns `None` here
840 /// simply never triggers an issuer-change warning.
841 async fn last_issuer(&self, _server_key: &str) -> Result<Option<String>> {
842 Ok(None)
843 }
844
845 /// Record the issuer currently in use for `server_key`.
846 ///
847 /// Defaults to `Ok(())` — a successful no-op — for the same reason
848 /// [`CredentialStore::last_issuer`] defaults to `None`.
849 async fn record_issuer(&self, _server_key: &str, _issuer: &str) -> Result<()> {
850 Ok(())
851 }
852}
853
854/// The operations an administrative tool needs, kept OFF the platform seam.
855///
856/// [`CredentialStore`] is what an OAuth flow needs and what a hosting platform
857/// implements. This trait is what a command-line tool needs in order to list,
858/// remove by server, wipe with an accurate count, and report on a migration —
859/// and a minimal platform store has no business implementing any of it. Giving
860/// those methods default bodies on the narrow seam would be worse than omitting
861/// them: a default `Ok(0)` is a lie that a tool would then print as a count.
862///
863/// So: `OAuthHelper` holds a [`CredentialStore`]; an administrative tool holds
864/// something that also implements this. A type that implements only
865/// [`CredentialStore`] does NOT satisfy this trait, which is the point — the
866/// platform seam stays narrow.
867///
868/// ```compile_fail
869/// use pmcp::CredentialStore;
870///
871/// // The narrow bound cannot reach the administrative operations.
872/// async fn wipe<S: CredentialStore>(store: &S) -> pmcp::Result<usize> {
873/// store.clear_all().await
874/// }
875/// ```
876#[async_trait]
877pub trait CredentialStoreAdmin: CredentialStore {
878 /// Every stored key, in a deterministic order.
879 async fn list_keys(&self) -> Result<Vec<CredentialKey>>;
880
881 /// Remove every credential whose key names `server_key`, returning how many
882 /// were removed. Removing zero is NOT an error.
883 async fn delete_by_server(&self, server_key: &str) -> Result<usize>;
884
885 /// Remove everything, returning how many credentials were removed.
886 async fn clear_all(&self) -> Result<usize>;
887
888 /// The report from the most recent load that performed a migration, if
889 /// there was one. Taking it CLEARS it, so an operator is told once.
890 async fn take_migration_report(&self) -> Result<Option<MigrationReport>>;
891}
892
893// ---------------------------------------------------------------------------
894// In-memory implementation
895// ---------------------------------------------------------------------------
896
897/// An in-memory [`CredentialStore`] and [`CredentialStoreAdmin`].
898///
899/// Everything is delegated to a [`CredentialSnapshot`] behind a lock, so this
900/// store and any document-backed store share ONE set of semantics and cannot
901/// drift apart.
902///
903/// # Examples
904///
905/// ```
906/// use pmcp::{CredentialStoreAdmin, InMemoryCredentialStore};
907///
908/// # async fn demo() -> pmcp::Result<()> {
909/// let legacy = br#"{"schema_version": 1, "entries": {}}"#;
910/// let store = InMemoryCredentialStore::from_bytes(legacy)?;
911/// assert_eq!(store.list_keys().await?.len(), 0);
912/// # Ok(())
913/// # }
914/// ```
915#[derive(Debug, Default)]
916pub struct InMemoryCredentialStore {
917 snapshot: RwLock<CredentialSnapshot>,
918 migration_report: RwLock<Option<MigrationReport>>,
919}
920
921impl InMemoryCredentialStore {
922 /// An empty store.
923 pub fn new() -> Self {
924 Self::default()
925 }
926
927 /// Seed a store from a serialized credential document, migrating it if
928 /// needed. Any resulting migration report is retained until it is taken via
929 /// [`CredentialStoreAdmin::take_migration_report`].
930 pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
931 let (snapshot, report) = parse_credential_snapshot(bytes)?;
932 let retained = if report.is_noop() { None } else { Some(report) };
933 Ok(Self {
934 snapshot: RwLock::new(snapshot),
935 migration_report: RwLock::new(retained),
936 })
937 }
938}
939
940#[async_trait]
941impl CredentialStore for InMemoryCredentialStore {
942 async fn load(&self, key: &CredentialKey) -> Result<Option<StoredCredentials>> {
943 Ok(self.snapshot.read().get(key).cloned())
944 }
945
946 async fn save(&self, key: &CredentialKey, credentials: &StoredCredentials) -> Result<()> {
947 self.snapshot
948 .write()
949 .insert(key.clone(), credentials.clone());
950 Ok(())
951 }
952
953 async fn delete(&self, key: &CredentialKey) -> Result<()> {
954 self.snapshot.write().remove(key);
955 Ok(())
956 }
957
958 /// Overridden to be atomic: both mutations happen under one write lock, so
959 /// no reader ever observes the credentials without the issuer record.
960 async fn save_with_issuer(
961 &self,
962 key: &CredentialKey,
963 credentials: &StoredCredentials,
964 server_key: &str,
965 issuer: &str,
966 ) -> Result<()> {
967 let mut snapshot = self.snapshot.write();
968 snapshot.insert(key.clone(), credentials.clone());
969 snapshot.record_issuer(server_key, issuer);
970 Ok(())
971 }
972
973 async fn last_issuer(&self, server_key: &str) -> Result<Option<String>> {
974 Ok(self
975 .snapshot
976 .read()
977 .last_issuer(server_key)
978 .map(str::to_owned))
979 }
980
981 async fn record_issuer(&self, server_key: &str, issuer: &str) -> Result<()> {
982 self.snapshot.write().record_issuer(server_key, issuer);
983 Ok(())
984 }
985}
986
987#[async_trait]
988impl CredentialStoreAdmin for InMemoryCredentialStore {
989 async fn list_keys(&self) -> Result<Vec<CredentialKey>> {
990 Ok(self.snapshot.read().keys())
991 }
992
993 async fn delete_by_server(&self, server_key: &str) -> Result<usize> {
994 let mut snapshot = self.snapshot.write();
995 let mut removed = 0usize;
996 for key in snapshot.keys_for_server(server_key) {
997 if snapshot.remove(&key) {
998 removed += 1;
999 }
1000 }
1001 snapshot.forget_issuer(server_key);
1002 Ok(removed)
1003 }
1004
1005 async fn clear_all(&self) -> Result<usize> {
1006 Ok(self.snapshot.write().clear())
1007 }
1008
1009 async fn take_migration_report(&self) -> Result<Option<MigrationReport>> {
1010 Ok(self.migration_report.write().take())
1011 }
1012}
1013
1014#[cfg(test)]
1015mod tests {
1016 use super::*;
1017
1018 #[test]
1019 fn the_redaction_marker_is_not_a_field_name_substring() {
1020 // Guards the redaction test's sentinel discipline: if the marker ever
1021 // became a field name, an "absence" assertion would pass vacuously.
1022 for field in [
1023 "access_token",
1024 "refresh_token",
1025 "expires_at",
1026 "scopes",
1027 "client_id",
1028 "registered_application_type",
1029 ] {
1030 assert!(!field.contains(REDACTED));
1031 assert!(!REDACTED.contains(field));
1032 }
1033 }
1034
1035 #[test]
1036 fn a_legacy_entry_carries_every_field_across() {
1037 let entry = LegacyEntry {
1038 access_token: "at".to_owned(),
1039 refresh_token: Some("rt".to_owned()),
1040 expires_at: Some(42),
1041 scopes: vec!["a".to_owned()],
1042 issuer: Some("https://as.example".to_owned()),
1043 client_id: "cid".to_owned(),
1044 };
1045 let stored = entry.into_stored();
1046 assert_eq!(stored.access_token(), "at");
1047 assert_eq!(stored.refresh_token(), Some("rt"));
1048 assert_eq!(stored.expires_at(), Some(42));
1049 assert_eq!(stored.granted_scopes(), ["a"]);
1050 assert_eq!(stored.client_id(), "cid");
1051 assert!(stored.registered_application_type().is_none());
1052 }
1053
1054 #[test]
1055 fn an_emptied_issuer_map_is_pruned_so_keys_stays_accurate() {
1056 let key = CredentialKey::new("https://as.example", "", "https://mcp.example");
1057 let mut snapshot = CredentialSnapshot::new();
1058 snapshot.insert(key.clone(), StoredCredentials::new("at", "cid"));
1059 assert!(snapshot.remove(&key));
1060 assert!(snapshot.credentials.is_empty(), "empty maps must be pruned");
1061 assert_eq!(snapshot.credential_count(), 0);
1062 }
1063
1064 #[test]
1065 fn forget_issuer_touches_only_the_named_server() {
1066 let mut snapshot = CredentialSnapshot::new();
1067 snapshot.record_issuer("https://a.example", "https://as.example");
1068 snapshot.record_issuer("https://b.example", "https://as.example");
1069 snapshot.forget_issuer("https://a.example");
1070 assert!(snapshot.last_issuer("https://a.example").is_none());
1071 assert!(snapshot.last_issuer("https://b.example").is_some());
1072 }
1073
1074 #[test]
1075 fn an_unsupported_version_refusal_names_both_versions() {
1076 let message = unsupported_schema_version(7).to_string();
1077 assert!(message.contains('7'), "{message}");
1078 assert!(message.contains('2'), "{message}");
1079 }
1080}