forked from
nonbinary.computer/jacquard
A better Rust ATProto crate
further cleanup
+117
-397
-15
crates/jacquard-common/src/session.rs
-15
crates/jacquard-common/src/session.rs
···
12
12
use std::path::{Path, PathBuf};
13
13
use std::sync::Arc;
14
14
use tokio::sync::RwLock;
15
-
use url::Url;
16
-
17
-
use crate::AuthorizationToken;
18
-
use crate::types::did::Did;
19
-
20
-
#[async_trait::async_trait]
21
-
pub trait Session {
22
-
async fn did(&self) -> Did<'_>;
23
-
24
-
async fn endpoint(&self) -> Url;
25
-
26
-
async fn access_token(&self) -> Result<AuthorizationToken, SessionStoreError>;
27
-
28
-
async fn refresh(&self) -> Result<AuthorizationToken, SessionStoreError>;
29
-
}
30
15
31
16
/// Errors emitted by session stores.
32
17
#[derive(Debug, thiserror::Error, Diagnostic)]
+71
-2
crates/jacquard-oauth/src/authstore.rs
+71
-2
crates/jacquard-oauth/src/authstore.rs
···
1
1
use std::sync::Arc;
2
2
3
+
use dashmap::DashMap;
3
4
use jacquard_common::{
4
5
IntoStatic,
5
-
session::{FileTokenStore, SessionStore, SessionStoreError},
6
+
session::{SessionStore, SessionStoreError},
6
7
types::did::Did,
7
8
};
8
-
use smol_str::SmolStr;
9
+
use smol_str::{SmolStr, ToSmolStr, format_smolstr};
9
10
10
11
use crate::session::{AuthRequestData, ClientSessionData};
11
12
···
37
38
) -> Result<(), SessionStoreError>;
38
39
39
40
async fn delete_auth_req_info(&self, state: &str) -> Result<(), SessionStoreError>;
41
+
}
42
+
43
+
pub struct MemoryAuthStore {
44
+
sessions: DashMap<SmolStr, ClientSessionData<'static>>,
45
+
auth_reqs: DashMap<SmolStr, AuthRequestData<'static>>,
46
+
}
47
+
48
+
impl MemoryAuthStore {
49
+
pub fn new() -> Self {
50
+
Self {
51
+
sessions: DashMap::new(),
52
+
auth_reqs: DashMap::new(),
53
+
}
54
+
}
55
+
}
56
+
57
+
#[async_trait::async_trait]
58
+
impl ClientAuthStore for MemoryAuthStore {
59
+
async fn get_session(
60
+
&self,
61
+
did: &Did<'_>,
62
+
session_id: &str,
63
+
) -> Result<Option<ClientSessionData<'_>>, SessionStoreError> {
64
+
let key = format_smolstr!("{}_{}", did, session_id);
65
+
Ok(self.sessions.get(&key).map(|v| v.clone()))
66
+
}
67
+
68
+
async fn upsert_session(
69
+
&self,
70
+
session: ClientSessionData<'_>,
71
+
) -> Result<(), SessionStoreError> {
72
+
let key = format_smolstr!("{}_{}", session.account_did, session.session_id);
73
+
self.sessions.insert(key, session.into_static());
74
+
Ok(())
75
+
}
76
+
77
+
async fn delete_session(
78
+
&self,
79
+
did: &Did<'_>,
80
+
session_id: &str,
81
+
) -> Result<(), SessionStoreError> {
82
+
let key = format_smolstr!("{}_{}", did, session_id);
83
+
self.sessions.remove(&key);
84
+
Ok(())
85
+
}
86
+
87
+
async fn get_auth_req_info(
88
+
&self,
89
+
state: &str,
90
+
) -> Result<Option<AuthRequestData<'_>>, SessionStoreError> {
91
+
Ok(self.auth_reqs.get(state).map(|v| v.clone()))
92
+
}
93
+
94
+
async fn save_auth_req_info(
95
+
&self,
96
+
auth_req_info: &AuthRequestData<'_>,
97
+
) -> Result<(), SessionStoreError> {
98
+
self.auth_reqs.insert(
99
+
auth_req_info.state.clone().to_smolstr(),
100
+
auth_req_info.clone().into_static(),
101
+
);
102
+
Ok(())
103
+
}
104
+
105
+
async fn delete_auth_req_info(&self, state: &str) -> Result<(), SessionStoreError> {
106
+
self.auth_reqs.remove(state);
107
+
Ok(())
108
+
}
40
109
}
41
110
42
111
#[async_trait::async_trait]
+8
crates/jacquard-oauth/src/client.rs
+8
crates/jacquard-oauth/src/client.rs
···
18
18
xrpc::{CallOptions, Response, XrpcClient, XrpcExt, XrpcRequest},
19
19
},
20
20
};
21
+
use jacquard_identity::JacquardResolver;
21
22
use jose_jwk::JwkSet;
22
23
use std::sync::Arc;
23
24
use tokio::sync::RwLock;
···
30
31
{
31
32
pub registry: Arc<SessionRegistry<T, S>>,
32
33
pub client: Arc<T>,
34
+
}
35
+
36
+
impl<S: ClientAuthStore> OAuthClient<JacquardResolver, S> {
37
+
pub fn new(store: S, client_data: ClientData<'static>) -> Self {
38
+
let client = JacquardResolver::default();
39
+
Self::new_from_resolver(store, client, client_data)
40
+
}
33
41
}
34
42
35
43
impl<T, S> OAuthClient<T, S>
+3
crates/jacquard-oauth/src/resolver.rs
+3
crates/jacquard-oauth/src/resolver.rs
+3
-60
crates/jacquard/src/client.rs
+3
-60
crates/jacquard/src/client.rs
···
3
3
//! This module provides HTTP and XRPC client traits along with an authenticated
4
4
//! client implementation that manages session tokens.
5
5
6
-
mod at_client;
7
6
pub mod credential_session;
8
-
mod token;
9
-
10
-
pub use at_client::{AtClient, SendOverrides};
7
+
pub mod token;
11
8
12
9
pub use jacquard_common::error::{ClientError, XrpcResult};
13
10
pub use jacquard_common::session::{MemorySessionStore, SessionStore, SessionStoreError};
14
11
use jacquard_common::{
15
12
CowStr, IntoStatic,
16
-
types::{
17
-
string::{Did, Handle},
18
-
xrpc::{Response, XrpcRequest},
19
-
},
13
+
types::string::{Did, Handle},
20
14
};
21
15
pub use token::FileAuthStore;
22
-
use url::Url;
23
16
24
17
pub(crate) const NSID_REFRESH_SESSION: &str = "com.atproto.server.refreshSession";
25
18
26
19
/// Basic client wrapper: reqwest transport + in-memory session store.
27
-
pub struct BasicClient(AtClient<reqwest::Client, MemorySessionStore<Did<'static>, AuthSession>>);
28
-
29
-
impl BasicClient {
30
-
/// Construct a basic client with minimal inputs.
31
-
pub fn new(base: Url) -> Self {
32
-
Self(AtClient::new(
33
-
reqwest::Client::new(),
34
-
base,
35
-
MemorySessionStore::default(),
36
-
))
37
-
}
38
-
39
-
/// Access the inner stateful client.
40
-
pub fn inner(
41
-
&self,
42
-
) -> &AtClient<reqwest::Client, MemorySessionStore<Did<'static>, AuthSession>> {
43
-
&self.0
44
-
}
45
-
46
-
/// Send an XRPC request.
47
-
pub async fn send<R: XrpcRequest + Send>(&self, req: R) -> XrpcResult<Response<R>> {
48
-
self.0.send(req).await
49
-
}
50
-
51
-
/// Send with per-call overrides.
52
-
pub async fn send_with<R: XrpcRequest + Send>(
53
-
&self,
54
-
req: R,
55
-
overrides: SendOverrides<'_>,
56
-
) -> XrpcResult<Response<R>> {
57
-
self.0.send_with(req, overrides).await
58
-
}
59
-
60
-
/// Get current session.
61
-
pub async fn session(&self, did: &Did<'static>) -> Option<AuthSession> {
62
-
self.0.session(did).await
63
-
}
64
-
65
-
/// Set the session.
66
-
pub async fn set_session(
67
-
&self,
68
-
session: AuthSession,
69
-
) -> core::result::Result<(), SessionStoreError> {
70
-
self.0.set_session(session).await
71
-
}
72
-
73
-
/// Base URL of this client.
74
-
pub fn base(&self) -> &Url {
75
-
self.0.base()
76
-
}
77
-
}
20
+
pub struct BasicClient(); //AtClient<reqwest::Client, MemorySessionStore<Did<'static>, AuthSession>>);
78
21
79
22
/// App password session information from `com.atproto.server.createSession`
80
23
///
-284
crates/jacquard/src/client/at_client.rs
-284
crates/jacquard/src/client/at_client.rs
···
1
-
use bytes::Bytes;
2
-
use jacquard_common::{
3
-
AuthorizationToken, IntoStatic,
4
-
error::{AuthError, ClientError, HttpError, TransportError, XrpcResult},
5
-
http_client::HttpClient,
6
-
session::{SessionStore, SessionStoreError},
7
-
types::{
8
-
did::Did,
9
-
xrpc::{CallOptions, Response, XrpcExt, XrpcRequest, build_http_request},
10
-
},
11
-
};
12
-
use url::Url;
13
-
14
-
use crate::client::{AtpSession, AuthSession, NSID_REFRESH_SESSION};
15
-
16
-
/// Per-call overrides when sending via `AtClient`.
17
-
#[derive(Debug, Clone)]
18
-
pub struct SendOverrides<'a> {
19
-
/// Optional DID override for this call.
20
-
pub did: Option<Did<'a>>,
21
-
/// Optional base URI override for this call.
22
-
pub base_uri: Option<Url>,
23
-
/// Per-request options such as auth, proxy, labelers, extra headers.
24
-
pub options: CallOptions<'a>,
25
-
/// Whether to auto-refresh on expired/invalid token and retry once.
26
-
pub auto_refresh: bool,
27
-
}
28
-
29
-
impl Default for SendOverrides<'_> {
30
-
fn default() -> Self {
31
-
Self {
32
-
did: None,
33
-
base_uri: None,
34
-
options: CallOptions::default(),
35
-
auto_refresh: true,
36
-
}
37
-
}
38
-
}
39
-
40
-
impl<'a> SendOverrides<'a> {
41
-
/// Construct default overrides (no base override, auto-refresh enabled).
42
-
pub fn new() -> Self {
43
-
Self {
44
-
did: None,
45
-
base_uri: None,
46
-
options: CallOptions::default(),
47
-
auto_refresh: true,
48
-
}
49
-
}
50
-
/// Override the base URI for this call only.
51
-
pub fn base_uri(mut self, base: Url) -> Self {
52
-
self.base_uri = Some(base);
53
-
self
54
-
}
55
-
/// Provide a full set of call options (auth/headers/etc.).
56
-
pub fn options(mut self, opts: CallOptions<'a>) -> Self {
57
-
self.options = opts;
58
-
self
59
-
}
60
-
61
-
/// Provide a full set of call options (auth/headers/etc.).
62
-
pub fn did(mut self, did: Did<'a>) -> Self {
63
-
self.did = Some(did);
64
-
self
65
-
}
66
-
/// Enable or disable one-shot auto-refresh + retry behavior.
67
-
pub fn auto_refresh(mut self, enable: bool) -> Self {
68
-
self.auto_refresh = enable;
69
-
self
70
-
}
71
-
}
72
-
73
-
/// Stateful client for AT Protocol XRPC with token storage and auto-refresh.
74
-
///
75
-
/// Example (file-backed tokens)
76
-
/// ```ignore
77
-
/// use jacquard::client::{AtClient, FileTokenStore, TokenStore};
78
-
/// use jacquard::api::com_atproto::server::create_session::CreateSession;
79
-
/// use jacquard::client::AtClient as _; // method resolution
80
-
/// use jacquard::CowStr;
81
-
///
82
-
/// #[tokio::main]
83
-
/// async fn main() -> miette::Result<()> {
84
-
/// let base = url::Url::parse("https://bsky.social")?;
85
-
/// let store = FileTokenStore::new("/tmp/jacquard-session.json");
86
-
/// let client = AtClient::new(reqwest::Client::new(), base, store);
87
-
/// let session = client
88
-
/// .send(
89
-
/// CreateSession::new()
90
-
/// .identifier(CowStr::from("alice.example"))
91
-
/// .password(CowStr::from("app-password"))
92
-
/// .build(),
93
-
/// )
94
-
/// .await?
95
-
/// .into_output()?;
96
-
/// client.set_session(session.into()).await?;
97
-
/// Ok(())
98
-
/// }
99
-
/// ```
100
-
pub struct AtClient<C: HttpClient, S> {
101
-
transport: C,
102
-
base: Url,
103
-
tokens: S,
104
-
refresh_lock: tokio::sync::Mutex<Option<Did<'static>>>,
105
-
}
106
-
107
-
impl<C: HttpClient, S: SessionStore<Did<'static>, AuthSession>> AtClient<C, S> {
108
-
/// Create a new client with a transport, base URL, and token store.
109
-
pub fn new(transport: C, base: Url, tokens: S) -> Self {
110
-
Self {
111
-
transport,
112
-
base,
113
-
tokens,
114
-
refresh_lock: tokio::sync::Mutex::new(None),
115
-
}
116
-
}
117
-
118
-
/// Get the base URL of this client.
119
-
pub fn base(&self) -> &Url {
120
-
&self.base
121
-
}
122
-
123
-
/// Access the underlying transport.
124
-
pub fn transport(&self) -> &C {
125
-
&self.transport
126
-
}
127
-
128
-
/// Get the current session, if any.
129
-
pub async fn session(&self, did: &Did<'static>) -> Option<AuthSession> {
130
-
self.tokens.get(did).await
131
-
}
132
-
133
-
/// Set the current session in the token store.
134
-
pub async fn set_session(&self, session: AuthSession) -> Result<(), SessionStoreError> {
135
-
let s = session.clone();
136
-
let did = s.did().clone().into_static();
137
-
self.refresh_lock.lock().await.replace(did.clone());
138
-
self.tokens.set(did, session).await
139
-
}
140
-
141
-
/// Send an XRPC request using the client's base URL and default behavior.
142
-
pub async fn send<R: XrpcRequest + Send>(&self, req: R) -> XrpcResult<Response<R>> {
143
-
self.send_with(req, SendOverrides::new()).await
144
-
}
145
-
146
-
/// Send an XRPC request with per-call overrides.
147
-
pub async fn send_with<R: XrpcRequest + Send>(
148
-
&self,
149
-
req: R,
150
-
mut overrides: SendOverrides<'_>,
151
-
) -> XrpcResult<Response<R>> {
152
-
let base = overrides
153
-
.base_uri
154
-
.clone()
155
-
.unwrap_or_else(|| self.base.clone());
156
-
let is_refresh = R::NSID == NSID_REFRESH_SESSION;
157
-
158
-
let mut current_did = None;
159
-
if overrides.options.auth.is_none() {
160
-
if let Ok(guard) = self.refresh_lock.try_lock() {
161
-
if let Some(ref did) = *guard {
162
-
current_did = Some(did.clone());
163
-
if let Some(s) = self.tokens.get(&did).await {
164
-
overrides.options.auth = Some(
165
-
if let Some(refresh_tok) = s.refresh_token()
166
-
&& is_refresh
167
-
{
168
-
AuthorizationToken::Bearer(refresh_tok.clone().into_static())
169
-
} else {
170
-
AuthorizationToken::Bearer(s.access_token().clone().into_static())
171
-
},
172
-
);
173
-
}
174
-
}
175
-
}
176
-
}
177
-
178
-
let http_request =
179
-
build_http_request(&base, &req, &overrides.options).map_err(TransportError::from)?;
180
-
let http_response = self
181
-
.transport
182
-
.send_http(http_request)
183
-
.await
184
-
.map_err(|e| TransportError::Other(Box::new(e)))?;
185
-
let status = http_response.status();
186
-
let buffer = Bytes::from(http_response.into_body());
187
-
188
-
if !status.is_success() && !matches!(status.as_u16(), 400 | 401) {
189
-
return Err(HttpError {
190
-
status,
191
-
body: Some(buffer),
192
-
}
193
-
.into());
194
-
}
195
-
196
-
if overrides.auto_refresh
197
-
&& !is_refresh
198
-
&& overrides.options.auth.is_some()
199
-
&& Self::is_auth_expired(status, &buffer)
200
-
{
201
-
self.refresh_once().await?;
202
-
203
-
let mut retry_opts = overrides.options.clone();
204
-
if let Some(curr_did) = current_did {
205
-
if let Some(s) = self.tokens.get(&curr_did).await {
206
-
retry_opts.auth = Some(AuthorizationToken::Bearer(
207
-
s.access_token().clone().into_static(),
208
-
));
209
-
}
210
-
}
211
-
let http_request =
212
-
build_http_request(&base, &req, &retry_opts).map_err(TransportError::from)?;
213
-
let http_response = self
214
-
.transport
215
-
.send_http(http_request)
216
-
.await
217
-
.map_err(|e| TransportError::Other(Box::new(e)))?;
218
-
let status = http_response.status();
219
-
let buffer = Bytes::from(http_response.into_body());
220
-
221
-
if !status.is_success() && !matches!(status.as_u16(), 400 | 401) {
222
-
return Err(HttpError {
223
-
status,
224
-
body: Some(buffer),
225
-
}
226
-
.into());
227
-
}
228
-
return Ok(Response::new(buffer, status));
229
-
}
230
-
231
-
Ok(Response::new(buffer, status))
232
-
}
233
-
234
-
async fn refresh_once(&self) -> XrpcResult<()> {
235
-
let guard = self.refresh_lock.lock().await;
236
-
if let Some(ref did) = *guard {
237
-
if let Some(s) = self.tokens.get(did).await {
238
-
if let Some(refresh_tok) = s.refresh_token() {
239
-
let refresh_resp = self
240
-
.transport
241
-
.xrpc(self.base.clone())
242
-
.auth(AuthorizationToken::Bearer(
243
-
refresh_tok.clone().into_static(),
244
-
))
245
-
.send(&jacquard_api::com_atproto::server::refresh_session::RefreshSession)
246
-
.await?;
247
-
let refreshed = match refresh_resp.into_output() {
248
-
Ok(o) => AtpSession::from(o),
249
-
Err(_) => return Err(ClientError::Auth(AuthError::RefreshFailed)),
250
-
};
251
-
252
-
let mut session = s.clone();
253
-
session.set_access_token(refreshed.access_jwt);
254
-
session.set_refresh_token(refreshed.refresh_jwt);
255
-
256
-
self.set_session(session)
257
-
.await
258
-
.map_err(|_| ClientError::Auth(AuthError::RefreshFailed))?;
259
-
Ok(())
260
-
} else {
261
-
Err(ClientError::Auth(AuthError::RefreshFailed))
262
-
}
263
-
} else {
264
-
Err(ClientError::Auth(AuthError::NotAuthenticated))
265
-
}
266
-
} else {
267
-
Err(ClientError::Auth(AuthError::NotAuthenticated))
268
-
}
269
-
}
270
-
271
-
fn is_auth_expired(status: http::StatusCode, buffer: &Bytes) -> bool {
272
-
if status.as_u16() == 401 {
273
-
return true;
274
-
}
275
-
if status.as_u16() == 400 {
276
-
if let Ok(val) = serde_json::from_slice::<serde_json::Value>(buffer) {
277
-
if let Some(code) = val.get("error").and_then(|v| v.as_str()) {
278
-
return matches!(code, "ExpiredToken" | "InvalidToken");
279
-
}
280
-
}
281
-
}
282
-
false
283
-
}
284
-
}
+3
-7
crates/jacquard/src/client/token.rs
+3
-7
crates/jacquard/src/client/token.rs
···
1
1
use jacquard_common::IntoStatic;
2
2
use jacquard_common::cowstr::ToCowStr;
3
3
use jacquard_common::session::{FileTokenStore, SessionStore, SessionStoreError};
4
-
use jacquard_common::types::string::{Datetime, Did, Handle};
4
+
use jacquard_common::types::string::{Datetime, Did};
5
5
use jacquard_oauth::scopes::Scope;
6
6
use jacquard_oauth::session::{AuthRequestData, ClientSessionData, DpopClientData, DpopReqData};
7
7
use jacquard_oauth::types::OAuthTokenType;
8
8
use jose_jwk::Key;
9
-
use serde::de::DeserializeOwned;
10
9
use serde::{Deserialize, Serialize};
11
10
use serde_json::Value;
12
-
use std::fmt::Display;
13
-
use std::hash::Hash;
14
-
use std::path::{Path, PathBuf};
15
11
use url::Url;
16
12
17
13
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
···
22
18
}
23
19
24
20
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
25
-
struct StoredAtSession {
21
+
pub struct StoredAtSession {
26
22
access_jwt: String,
27
23
refresh_jwt: String,
28
24
did: String,
···
32
28
}
33
29
34
30
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
35
-
struct OAuthSession {
31
+
pub struct OAuthSession {
36
32
account_did: String,
37
33
session_id: String,
38
34
+29
-29
crates/jacquard/src/main.rs
+29
-29
crates/jacquard/src/main.rs
···
31
31
let resolver = slingshot_resolver_default();
32
32
let handle = Handle::new(args.username.as_ref()).into_diagnostic()?;
33
33
let (_did, pds_url) = resolver.pds_for_handle(&handle).await.into_diagnostic()?;
34
-
let client = BasicClient::new(pds_url);
34
+
// let client = BasicClient::new(pds_url);
35
35
36
-
// Create session
37
-
let session = AtpSession::from(
38
-
client
39
-
.send(
40
-
CreateSession::new()
41
-
.identifier(args.username)
42
-
.password(args.password)
43
-
.build(),
44
-
)
45
-
.await?
46
-
.into_output()?,
47
-
);
36
+
// // Create session
37
+
// let session = AtpSession::from(
38
+
// client
39
+
// .send(
40
+
// CreateSession::new()
41
+
// .identifier(args.username)
42
+
// .password(args.password)
43
+
// .build(),
44
+
// )
45
+
// .await?
46
+
// .into_output()?,
47
+
// );
48
48
49
-
println!("logged in as {} ({})", session.handle, session.did);
50
-
client.set_session(session.into()).await.into_diagnostic()?;
49
+
// println!("logged in as {} ({})", session.handle, session.did);
50
+
// client.set_session(session.into()).await.into_diagnostic()?;
51
51
52
-
// Fetch timeline
53
-
println!("\nfetching timeline...");
54
-
let timeline = client
55
-
.send(GetTimeline::new().limit(5).build())
56
-
.await?
57
-
.into_output()?;
52
+
// // Fetch timeline
53
+
// println!("\nfetching timeline...");
54
+
// let timeline = client
55
+
// .send(GetTimeline::new().limit(5).build())
56
+
// .await?
57
+
// .into_output()?;
58
58
59
-
println!("\ntimeline ({} posts):", timeline.feed.len());
60
-
for (i, post) in timeline.feed.iter().enumerate() {
61
-
println!("\n{}. by {}", i + 1, post.post.author.handle);
62
-
println!(
63
-
" {}",
64
-
serde_json::to_string_pretty(&post.post.record).into_diagnostic()?
65
-
);
66
-
}
59
+
// println!("\ntimeline ({} posts):", timeline.feed.len());
60
+
// for (i, post) in timeline.feed.iter().enumerate() {
61
+
// println!("\n{}. by {}", i + 1, post.post.author.handle);
62
+
// println!(
63
+
// " {}",
64
+
// serde_json::to_string_pretty(&post.post.record).into_diagnostic()?
65
+
// );
66
+
// }
67
67
68
68
Ok(())
69
69
}