1#![doc = include_str!("../README.md")]
2#![forbid(unsafe_code)]
3
4mod handlers;
5pub mod presets;
6mod seal;
7mod service;
8mod store;
9
10use std::sync::{Arc, OnceLock};
11
12use async_trait::async_trait;
13use sha2::{Digest, Sha256};
14use time::Duration;
15use zeroize::Zeroizing;
16
17use cratefield_core::{
18 Action, AnyError, BoxFuture, Config, ConfigError, DataKind, Disposition, Migrations, Module,
19 ModuleContext, Outcome, PersonalDataSet, Port, Problem, RoutePolicy, SqlMigration, Surface,
20};
21
22use cratefield_oauth_client::{ClientAuth, OAuthError};
23use handlers::Settings;
24
25const MIGRATION_INIT: SqlMigration = SqlMigration::new(
30 "0001",
31 "init",
32 include_str!("../migrations/sqlite/0001_init.sql"),
33);
34
35#[must_use]
39pub(crate) fn sha256_hex(value: &str) -> String {
40 hex::encode(Sha256::digest(value.as_bytes()))
41}
42
43#[must_use]
49pub(crate) fn origin_of(url: &str) -> Option<String> {
50 let (scheme, rest) = url.split_once("://")?;
51 let scheme = scheme.to_ascii_lowercase();
52 if scheme != "http" && scheme != "https" {
53 return None;
54 }
55 let authority = rest
56 .split(['/', '?', '#'])
57 .next()
58 .filter(|authority| !authority.is_empty())?;
59 if authority.contains('@') {
60 return None;
63 }
64 let authority = authority.to_ascii_lowercase();
65 let default_port = if scheme == "https" { ":443" } else { ":80" };
66 let host = authority.strip_suffix(default_port).unwrap_or(&authority);
67 Some(format!("{scheme}://{host}"))
68}
69
70pub(crate) fn encode_pair(name: &str, value: &str) -> String {
77 use std::fmt::Write as _;
78 fn encode(text: &str) -> String {
79 let mut out = String::with_capacity(text.len());
80 for byte in text.bytes() {
81 match byte {
82 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
83 out.push(byte as char);
84 }
85 _ => {
86 let _ = write!(out, "%{byte:02X}");
87 }
88 }
89 }
90 out
91 }
92 format!("{}={}", encode(name), encode(value))
93}
94
95#[must_use]
102pub(crate) fn safe_token(text: &str) -> &str {
103 let tokenish = !text.is_empty()
104 && text.len() <= 64
105 && text
106 .bytes()
107 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'));
108 if tokenish { text } else { "other" }
109}
110
111#[must_use]
114pub(crate) fn append_query(url: &str, name: &str, value: &str) -> String {
115 let pair = encode_pair(name, value);
116 let (head, fragment) = url
117 .split_once('#')
118 .map_or((url, ""), |(head, fragment)| (head, fragment));
119 let separator = if head.contains('?') { '&' } else { '?' };
120 if fragment.is_empty() {
121 format!("{head}{separator}{pair}")
122 } else {
123 format!("{head}{separator}{pair}#{fragment}")
124 }
125}
126
127#[derive(Debug, Clone, PartialEq, Eq)]
138pub struct Provider {
139 pub(crate) key: String,
140 pub(crate) authorize_url: String,
141 pub(crate) token_url: String,
142 pub(crate) revoke_url: Option<String>,
143 pub(crate) scopes: Vec<String>,
144 pub(crate) scope_separator: &'static str,
145 pub(crate) client_auth: ClientAuth,
146 pub(crate) authorize_params: Vec<(String, String)>,
147 pub(crate) rotating_refresh: bool,
148 pub(crate) pkce: bool,
149}
150
151impl Provider {
152 #[must_use]
156 pub fn new(
157 key: impl Into<String>,
158 authorize_url: impl Into<String>,
159 token_url: impl Into<String>,
160 ) -> Self {
161 Self {
162 key: key.into(),
163 authorize_url: authorize_url.into(),
164 token_url: token_url.into(),
165 revoke_url: None,
166 scopes: Vec::new(),
167 scope_separator: " ",
168 client_auth: ClientAuth::ClientSecretPost,
169 authorize_params: Vec::new(),
170 rotating_refresh: false,
171 pkce: false,
172 }
173 }
174
175 #[must_use]
179 pub fn key(&self) -> &str {
180 &self.key
181 }
182
183 #[must_use]
186 pub fn with_scopes<I, S>(mut self, scopes: I) -> Self
187 where
188 I: IntoIterator<Item = S>,
189 S: Into<String>,
190 {
191 self.scopes = scopes.into_iter().map(Into::into).collect();
192 self
193 }
194
195 #[must_use]
197 pub fn with_revoke_url(mut self, url: impl Into<String>) -> Self {
198 self.revoke_url = Some(url.into());
199 self
200 }
201
202 #[must_use]
205 pub fn with_scope_separator(mut self, separator: &'static str) -> Self {
206 self.scope_separator = separator;
207 self
208 }
209
210 #[must_use]
212 pub fn with_client_auth(mut self, auth: ClientAuth) -> Self {
213 self.client_auth = auth;
214 self
215 }
216
217 #[must_use]
220 pub fn with_authorize_param(
221 mut self,
222 name: impl Into<String>,
223 value: impl Into<String>,
224 ) -> Self {
225 self.authorize_params.push((name.into(), value.into()));
226 self
227 }
228
229 #[must_use]
232 pub fn with_rotating_refresh(mut self, rotating: bool) -> Self {
233 self.rotating_refresh = rotating;
234 self
235 }
236
237 #[must_use]
239 pub fn with_pkce(mut self, pkce: bool) -> Self {
240 self.pkce = pkce;
241 self
242 }
243
244 pub(crate) fn env_key(&self) -> String {
246 service::env_key(&self.key)
247 }
248
249 #[must_use]
253 pub fn rotates_refresh(&self) -> bool {
254 self.rotating_refresh
255 }
256}
257
258#[derive(Debug, Clone, Copy, PartialEq, Eq)]
263#[non_exhaustive]
264pub enum ConnectionStatus {
265 Active,
267 NeedsReconnect,
269 Revoked,
271}
272
273impl ConnectionStatus {
274 #[must_use]
276 pub fn as_str(self) -> &'static str {
277 match self {
278 Self::Active => "active",
279 Self::NeedsReconnect => "needs_reconnect",
280 Self::Revoked => "revoked",
281 }
282 }
283
284 #[must_use]
288 pub(crate) fn parse(raw: &str) -> Self {
289 match raw {
290 "active" => Self::Active,
291 "revoked" => Self::Revoked,
292 _ => Self::NeedsReconnect,
293 }
294 }
295}
296
297#[derive(Debug, Clone, PartialEq, Eq)]
301pub struct Connection {
302 pub id: String,
304 pub subject: String,
306 pub provider: String,
308 pub external_account_id: Option<String>,
310 pub display_name: Option<String>,
312 pub scopes: Vec<String>,
314 pub status: ConnectionStatus,
316 pub access_expires_at: Option<String>,
319 pub refresh_expires_at: Option<String>,
321 pub last_error: Option<String>,
323 pub created_at: String,
325 pub updated_at: String,
327}
328
329impl From<store::ConnectionRow> for Connection {
330 fn from(row: store::ConnectionRow) -> Self {
331 Self {
332 id: row.id,
333 subject: row.subject,
334 provider: row.provider,
335 external_account_id: row.external_account_id,
336 display_name: row.display_name,
337 scopes: store::scopes_of(&row.scopes),
338 status: ConnectionStatus::parse(&row.status),
339 access_expires_at: row.access_expires_at,
340 refresh_expires_at: row.refresh_expires_at,
341 last_error: row.last_error,
342 created_at: row.created_at,
343 updated_at: row.updated_at,
344 }
345 }
346}
347
348pub struct AccessToken {
353 connection_id: String,
354 token: Zeroizing<String>,
355 expires_at: Option<String>,
356}
357
358impl AccessToken {
359 pub(crate) fn new(
360 connection_id: String,
361 token: Zeroizing<String>,
362 expires_at: Option<String>,
363 ) -> Self {
364 Self {
365 connection_id,
366 token,
367 expires_at,
368 }
369 }
370
371 #[must_use]
373 pub fn connection_id(&self) -> &str {
374 &self.connection_id
375 }
376
377 #[must_use]
380 pub fn expose(&self) -> &str {
381 &self.token
382 }
383
384 #[must_use]
387 pub fn into_secret(self) -> Zeroizing<String> {
388 self.token
389 }
390
391 #[must_use]
393 pub fn expires_at(&self) -> Option<&str> {
394 self.expires_at.as_deref()
395 }
396}
397
398impl std::fmt::Debug for AccessToken {
399 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
400 f.debug_struct("AccessToken")
401 .field("connection_id", &self.connection_id)
402 .field("token", &"<redacted>")
403 .field("expires_at", &self.expires_at)
404 .finish()
405 }
406}
407
408#[derive(Debug, Clone, PartialEq, Eq)]
411pub struct AuthorizeUrl {
412 pub url: String,
414 pub expires_at: String,
416 pub provider: String,
418}
419
420#[derive(Debug, Clone, PartialEq, Eq)]
427pub struct ProviderFailure {
428 pub status: Option<u16>,
431 pub code: String,
435}
436
437impl ProviderFailure {
438 #[must_use]
440 pub fn of(error: &OAuthError) -> Self {
441 match error {
442 OAuthError::Provider(inner) => Self {
443 status: Some(inner.status),
444 code: safe_token(&inner.code).to_owned(),
445 },
446 OAuthError::Transport(_) => Self {
447 status: None,
448 code: "transport".to_owned(),
449 },
450 OAuthError::Decode(_) => Self {
451 status: None,
452 code: "decode".to_owned(),
453 },
454 OAuthError::Config(_) => Self {
455 status: None,
456 code: "config".to_owned(),
457 },
458 OAuthError::NoRevokeEndpoint => Self {
459 status: None,
460 code: "no_revoke_endpoint".to_owned(),
461 },
462 }
463 }
464}
465
466impl std::fmt::Display for ProviderFailure {
467 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
468 match self.status {
469 Some(status) => write!(f, "the provider answered {status} (error `{}`)", self.code),
470 None => write!(f, "the provider call failed ({})", self.code),
471 }
472 }
473}
474
475impl std::error::Error for ProviderFailure {}
476
477#[derive(Debug, thiserror::Error)]
479#[non_exhaustive]
480pub enum ConnectionError {
481 #[error("`{0}` is not a configured provider")]
483 UnknownProvider(String),
484 #[error("`{0}` is not on an allowed origin")]
486 OriginNotAllowed(String),
487 #[error("the connect state is unknown, spent or expired")]
489 BadState,
490 #[error("no connection with id `{0}`")]
492 NotFound(String),
493 #[error("the connection needs to be reconnected: {0}")]
495 NeedsReconnect(String),
496 #[error("the connection was revoked")]
498 Revoked,
499 #[error(transparent)]
501 Provider(ProviderFailure),
502 #[error("connections is misconfigured: {0}")]
504 Config(String),
505 #[error(transparent)]
507 Db(#[from] cratefield_core::DbError),
508}
509
510impl ConnectionError {
511 #[must_use]
514 pub fn slug(&self) -> &'static str {
515 handlers::problem_for(self).slug
516 }
517
518 #[must_use]
520 pub fn problem(&self) -> Problem {
521 handlers::problem_for(self)
522 }
523
524 #[must_use]
527 pub fn provider_failure(&self) -> Option<&ProviderFailure> {
528 match self {
529 Self::Provider(failure) => Some(failure),
530 _ => None,
531 }
532 }
533}
534
535pub struct Connections {
542 settings: Settings,
543 ctx: Arc<OnceLock<Arc<ModuleContext>>>,
547}
548
549impl Default for Connections {
550 fn default() -> Self {
551 Self::builder().build()
552 }
553}
554
555impl Connections {
556 #[must_use]
558 pub fn builder() -> ConnectionsBuilder {
559 ConnectionsBuilder::new()
560 }
561
562 #[must_use]
566 pub fn api(&self) -> ConnectionsApi {
567 ConnectionsApi {
568 ctx: Arc::clone(&self.ctx),
569 settings: self.settings.clone(),
570 }
571 }
572}
573
574pub struct ConnectionsBuilder {
576 settings: Settings,
577}
578
579impl ConnectionsBuilder {
580 fn new() -> Self {
581 Self {
582 settings: Settings {
583 providers: Vec::new(),
584 allowed_origins: Vec::new(),
585 refresh_lead_secs: service::DEFAULT_REFRESH_LEAD_SECS,
586 },
587 }
588 }
589
590 #[must_use]
593 pub fn provider(mut self, provider: Provider) -> Self {
594 self.settings
595 .providers
596 .retain(|existing| existing.key != provider.key);
597 self.settings.providers.push(provider);
598 self
599 }
600
601 #[must_use]
603 pub fn providers<I>(mut self, providers: I) -> Self
604 where
605 I: IntoIterator<Item = Provider>,
606 {
607 for provider in providers {
608 self = self.provider(provider);
609 }
610 self
611 }
612
613 #[must_use]
627 pub fn allowed_origin(mut self, origin: impl AsRef<str>) -> Self {
628 let raw = origin.as_ref();
629 let normalized = origin_of(raw).unwrap_or_else(|| {
630 panic!("connections: `{raw}` is not an absolute http(s) origin, e.g. https://app.example.com")
631 });
632 if !self.settings.allowed_origins.contains(&normalized) {
633 self.settings.allowed_origins.push(normalized);
634 }
635 self
636 }
637
638 #[must_use]
641 pub fn refresh_lead(mut self, lead: Duration) -> Self {
642 self.settings.refresh_lead_secs = lead.whole_seconds().max(0);
643 self
644 }
645
646 #[must_use]
648 pub fn build(self) -> Connections {
649 Connections {
650 settings: self.settings,
651 ctx: Arc::new(OnceLock::new()),
652 }
653 }
654}
655
656#[derive(Clone)]
659pub struct ConnectionsApi {
660 ctx: Arc<OnceLock<Arc<ModuleContext>>>,
661 settings: Settings,
662}
663
664impl ConnectionsApi {
665 fn ctx(&self) -> Result<Arc<ModuleContext>, ConnectionError> {
666 self.ctx.get().cloned().ok_or_else(|| {
667 ConnectionError::Config(
668 "the connections module has not been mounted yet; build the harness first"
669 .to_owned(),
670 )
671 })
672 }
673
674 pub async fn start(
679 &self,
680 subject: &str,
681 provider: &str,
682 return_to: &str,
683 ) -> Result<AuthorizeUrl, ConnectionError> {
684 service::start(
685 self.ctx()?.as_ref(),
686 &self.settings,
687 subject,
688 provider,
689 return_to,
690 )
691 .await
692 }
693
694 pub async fn complete(&self, state: &str, code: &str) -> Result<Connection, ConnectionError> {
699 service::complete(self.ctx()?.as_ref(), &self.settings, state, code).await
700 }
701
702 pub async fn list(&self, subject: &str) -> Result<Vec<Connection>, ConnectionError> {
706 service::list(self.ctx()?.as_ref(), subject).await
707 }
708
709 pub async fn get(&self, connection_id: &str) -> Result<Connection, ConnectionError> {
713 service::get(self.ctx()?.as_ref(), connection_id).await
714 }
715
716 pub async fn access_token(&self, connection_id: &str) -> Result<AccessToken, ConnectionError> {
720 service::access_token(self.ctx()?.as_ref(), &self.settings, connection_id).await
721 }
722
723 pub async fn revoke(&self, connection_id: &str) -> Result<(), ConnectionError> {
728 service::revoke(self.ctx()?.as_ref(), &self.settings, connection_id).await
729 }
730
731 pub async fn mark_needs_reconnect(
735 &self,
736 connection_id: &str,
737 reason: &str,
738 ) -> Result<(), ConnectionError> {
739 service::mark_needs_reconnect(self.ctx()?.as_ref(), connection_id, reason).await
740 }
741
742 pub async fn set_account(
747 &self,
748 connection_id: &str,
749 external_account_id: Option<&str>,
750 display_name: Option<&str>,
751 ) -> Result<(), ConnectionError> {
752 service::set_account(
753 self.ctx()?.as_ref(),
754 connection_id,
755 external_account_id,
756 display_name,
757 )
758 .await
759 }
760}
761
762#[async_trait]
766impl Module for Connections {
767 fn name(&self) -> &'static str {
768 "connections"
769 }
770
771 fn version(&self) -> &'static str {
772 env!("CARGO_PKG_VERSION")
773 }
774
775 fn requires(&self) -> &'static [Port] {
779 &[Port::Db, Port::HttpClient]
780 }
781
782 fn optional(&self) -> &'static [Port] {
786 &[Port::Clock, Port::IdGen, Port::Defer]
787 }
788
789 fn tables(&self) -> &'static [&'static str] {
790 &["connection", "connection_state"]
791 }
792
793 fn personal_data(&self) -> &'static [PersonalDataSet] {
801 static SETS: OnceLock<Vec<PersonalDataSet>> = OnceLock::new();
802 SETS.get_or_init(|| {
803 vec![
804 PersonalDataSet {
805 table: "connection",
806 subject: "subject",
807 kind: DataKind::Identifier,
808 disposition: Disposition::Erase,
809 description: "A third-party account you connected: the provider, the \
810 account id and name it gave us, the scopes, and the live status. The \
811 access and refresh tokens are held only as ciphertext and are never \
812 copied into an export or a response.",
813 redacted: &["access_token_sealed", "refresh_token_sealed"],
814 subject_via: None,
815 },
816 PersonalDataSet {
817 table: "connection_state",
818 subject: "subject",
819 kind: DataKind::Identifier,
820 disposition: Disposition::Erase,
821 description: "A connection attempt still in flight: where the browser was \
822 sent and when it lapses. The anti-forgery state is held only as a \
823 SHA-256 hash and the PKCE verifier only as ciphertext. Once spent, or \
824 ten minutes old, the row is deleted.",
825 redacted: &["state_hash", "verifier_sealed"],
826 subject_via: None,
827 },
828 ]
829 })
830 .as_slice()
831 }
832
833 fn emits(&self) -> &'static [&'static str] {
836 &[
837 handlers::EVENT_CONNECTED,
838 handlers::EVENT_REFRESHED,
839 handlers::EVENT_REVOKED,
840 handlers::EVENT_NEEDS_RECONNECT,
841 ]
842 }
843
844 fn migrations(&self) -> Migrations {
845 const MIGRATIONS: [SqlMigration; 1] = [MIGRATION_INIT];
846 const _: () = cratefield_core::assert_migration_set(&MIGRATIONS);
847 Migrations::sqlite(&MIGRATIONS)
848 }
849
850 fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
855 handlers::validate(&self.settings, cfg).into_result()
856 }
857
858 fn self_check(&self) -> Vec<String> {
861 handlers::self_check(&self.settings)
862 }
863
864 fn router(&self, ctx: ModuleContext) -> axum::Router {
865 let ctx = Arc::new(ctx);
866 let _ = self.ctx.set(Arc::clone(&ctx));
870 handlers::router(ctx, self.settings.clone())
871 }
872
873 fn surface(&self) -> Surface {
882 Surface::new().action(
883 Action::get("callback", "/callback/{provider}")
884 .policy(RoutePolicy::Open)
885 .outcome(Outcome::Redirect),
886 )
887 }
888
889 fn scheduled<'a>(
893 &'a self,
894 ctx: &'a ModuleContext,
895 cron: &'a str,
896 ) -> BoxFuture<'a, Result<(), AnyError>> {
897 Box::pin(async move {
898 if let Err(error) = service::maintain(ctx, &self.settings).await {
899 let failure = error.provider_failure();
900 tracing::warn!(
901 cron,
902 slug = error.slug(),
903 status = ?failure.and_then(|failure| failure.status),
904 code = failure.map_or("-", |failure| failure.code.as_str()),
905 "the scheduled connections pass did not complete"
906 );
907 }
908 Ok(())
909 })
910 }
911}
912
913#[cfg(test)]
914mod tests {
915 use super::*;
916
917 #[test]
920 fn a_query_pair_goes_before_the_fragment() {
921 assert_eq!(
922 append_query("https://app.example.com/settings#tab", "connection", "c-1"),
923 "https://app.example.com/settings?connection=c-1#tab"
924 );
925 assert_eq!(
926 append_query("https://app.example.com/settings?a=1#tab", "b", "2"),
927 "https://app.example.com/settings?a=1&b=2#tab"
928 );
929 assert_eq!(
930 append_query("https://app.example.com/s", "b", "2"),
931 "https://app.example.com/s?b=2"
932 );
933 }
934
935 #[test]
938 fn only_a_short_token_survives_sanitising() {
939 assert_eq!(safe_token("access_denied"), "access_denied");
940 assert_eq!(safe_token("invalid_grant"), "invalid_grant");
941 assert_eq!(safe_token("line\nbreak"), "other");
942 assert_eq!(safe_token(""), "other");
943 assert_eq!(safe_token(&"x".repeat(65)), "other");
944 }
945
946 #[test]
949 #[should_panic(expected = "is not an absolute http(s) origin")]
950 fn a_bad_allowed_origin_panics() {
951 let _ = Connections::builder().allowed_origin("app.example.com");
952 }
953}