#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
mod handlers;
pub mod presets;
mod seal;
mod service;
mod store;
use std::sync::{Arc, OnceLock};
use async_trait::async_trait;
use sha2::{Digest, Sha256};
use time::Duration;
use zeroize::Zeroizing;
use cratefield_core::{
Action, AnyError, BoxFuture, Config, ConfigError, DataKind, Disposition, Migrations, Module,
ModuleContext, Outcome, PersonalDataSet, Port, Problem, RoutePolicy, SqlMigration, Surface,
};
use cratefield_oauth_client::{ClientAuth, OAuthError};
use handlers::Settings;
const MIGRATION_INIT: SqlMigration = SqlMigration::new(
"0001",
"init",
include_str!("../migrations/sqlite/0001_init.sql"),
);
#[must_use]
pub(crate) fn sha256_hex(value: &str) -> String {
hex::encode(Sha256::digest(value.as_bytes()))
}
#[must_use]
pub(crate) fn origin_of(url: &str) -> Option<String> {
let (scheme, rest) = url.split_once("://")?;
let scheme = scheme.to_ascii_lowercase();
if scheme != "http" && scheme != "https" {
return None;
}
let authority = rest
.split(['/', '?', '#'])
.next()
.filter(|authority| !authority.is_empty())?;
if authority.contains('@') {
return None;
}
let authority = authority.to_ascii_lowercase();
let default_port = if scheme == "https" { ":443" } else { ":80" };
let host = authority.strip_suffix(default_port).unwrap_or(&authority);
Some(format!("{scheme}://{host}"))
}
pub(crate) fn encode_pair(name: &str, value: &str) -> String {
use std::fmt::Write as _;
fn encode(text: &str) -> String {
let mut out = String::with_capacity(text.len());
for byte in text.bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(byte as char);
}
_ => {
let _ = write!(out, "%{byte:02X}");
}
}
}
out
}
format!("{}={}", encode(name), encode(value))
}
#[must_use]
pub(crate) fn safe_token(text: &str) -> &str {
let tokenish = !text.is_empty()
&& text.len() <= 64
&& text
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'));
if tokenish { text } else { "other" }
}
#[must_use]
pub(crate) fn append_query(url: &str, name: &str, value: &str) -> String {
let pair = encode_pair(name, value);
let (head, fragment) = url
.split_once('#')
.map_or((url, ""), |(head, fragment)| (head, fragment));
let separator = if head.contains('?') { '&' } else { '?' };
if fragment.is_empty() {
format!("{head}{separator}{pair}")
} else {
format!("{head}{separator}{pair}#{fragment}")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Provider {
pub(crate) key: String,
pub(crate) authorize_url: String,
pub(crate) token_url: String,
pub(crate) revoke_url: Option<String>,
pub(crate) scopes: Vec<String>,
pub(crate) scope_separator: &'static str,
pub(crate) client_auth: ClientAuth,
pub(crate) authorize_params: Vec<(String, String)>,
pub(crate) rotating_refresh: bool,
pub(crate) pkce: bool,
}
impl Provider {
#[must_use]
pub fn new(
key: impl Into<String>,
authorize_url: impl Into<String>,
token_url: impl Into<String>,
) -> Self {
Self {
key: key.into(),
authorize_url: authorize_url.into(),
token_url: token_url.into(),
revoke_url: None,
scopes: Vec::new(),
scope_separator: " ",
client_auth: ClientAuth::ClientSecretPost,
authorize_params: Vec::new(),
rotating_refresh: false,
pkce: false,
}
}
#[must_use]
pub fn key(&self) -> &str {
&self.key
}
#[must_use]
pub fn with_scopes<I, S>(mut self, scopes: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.scopes = scopes.into_iter().map(Into::into).collect();
self
}
#[must_use]
pub fn with_revoke_url(mut self, url: impl Into<String>) -> Self {
self.revoke_url = Some(url.into());
self
}
#[must_use]
pub fn with_scope_separator(mut self, separator: &'static str) -> Self {
self.scope_separator = separator;
self
}
#[must_use]
pub fn with_client_auth(mut self, auth: ClientAuth) -> Self {
self.client_auth = auth;
self
}
#[must_use]
pub fn with_authorize_param(
mut self,
name: impl Into<String>,
value: impl Into<String>,
) -> Self {
self.authorize_params.push((name.into(), value.into()));
self
}
#[must_use]
pub fn with_rotating_refresh(mut self, rotating: bool) -> Self {
self.rotating_refresh = rotating;
self
}
#[must_use]
pub fn with_pkce(mut self, pkce: bool) -> Self {
self.pkce = pkce;
self
}
pub(crate) fn env_key(&self) -> String {
service::env_key(&self.key)
}
#[must_use]
pub fn rotates_refresh(&self) -> bool {
self.rotating_refresh
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ConnectionStatus {
Active,
NeedsReconnect,
Revoked,
}
impl ConnectionStatus {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::Active => "active",
Self::NeedsReconnect => "needs_reconnect",
Self::Revoked => "revoked",
}
}
#[must_use]
pub(crate) fn parse(raw: &str) -> Self {
match raw {
"active" => Self::Active,
"revoked" => Self::Revoked,
_ => Self::NeedsReconnect,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Connection {
pub id: String,
pub subject: String,
pub provider: String,
pub external_account_id: Option<String>,
pub display_name: Option<String>,
pub scopes: Vec<String>,
pub status: ConnectionStatus,
pub access_expires_at: Option<String>,
pub refresh_expires_at: Option<String>,
pub last_error: Option<String>,
pub created_at: String,
pub updated_at: String,
}
impl From<store::ConnectionRow> for Connection {
fn from(row: store::ConnectionRow) -> Self {
Self {
id: row.id,
subject: row.subject,
provider: row.provider,
external_account_id: row.external_account_id,
display_name: row.display_name,
scopes: store::scopes_of(&row.scopes),
status: ConnectionStatus::parse(&row.status),
access_expires_at: row.access_expires_at,
refresh_expires_at: row.refresh_expires_at,
last_error: row.last_error,
created_at: row.created_at,
updated_at: row.updated_at,
}
}
}
pub struct AccessToken {
connection_id: String,
token: Zeroizing<String>,
expires_at: Option<String>,
}
impl AccessToken {
pub(crate) fn new(
connection_id: String,
token: Zeroizing<String>,
expires_at: Option<String>,
) -> Self {
Self {
connection_id,
token,
expires_at,
}
}
#[must_use]
pub fn connection_id(&self) -> &str {
&self.connection_id
}
#[must_use]
pub fn expose(&self) -> &str {
&self.token
}
#[must_use]
pub fn into_secret(self) -> Zeroizing<String> {
self.token
}
#[must_use]
pub fn expires_at(&self) -> Option<&str> {
self.expires_at.as_deref()
}
}
impl std::fmt::Debug for AccessToken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AccessToken")
.field("connection_id", &self.connection_id)
.field("token", &"<redacted>")
.field("expires_at", &self.expires_at)
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthorizeUrl {
pub url: String,
pub expires_at: String,
pub provider: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderFailure {
pub status: Option<u16>,
pub code: String,
}
impl ProviderFailure {
#[must_use]
pub fn of(error: &OAuthError) -> Self {
match error {
OAuthError::Provider(inner) => Self {
status: Some(inner.status),
code: safe_token(&inner.code).to_owned(),
},
OAuthError::Transport(_) => Self {
status: None,
code: "transport".to_owned(),
},
OAuthError::Decode(_) => Self {
status: None,
code: "decode".to_owned(),
},
OAuthError::Config(_) => Self {
status: None,
code: "config".to_owned(),
},
OAuthError::NoRevokeEndpoint => Self {
status: None,
code: "no_revoke_endpoint".to_owned(),
},
}
}
}
impl std::fmt::Display for ProviderFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.status {
Some(status) => write!(f, "the provider answered {status} (error `{}`)", self.code),
None => write!(f, "the provider call failed ({})", self.code),
}
}
}
impl std::error::Error for ProviderFailure {}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ConnectionError {
#[error("`{0}` is not a configured provider")]
UnknownProvider(String),
#[error("`{0}` is not on an allowed origin")]
OriginNotAllowed(String),
#[error("the connect state is unknown, spent or expired")]
BadState,
#[error("no connection with id `{0}`")]
NotFound(String),
#[error("the connection needs to be reconnected: {0}")]
NeedsReconnect(String),
#[error("the connection was revoked")]
Revoked,
#[error(transparent)]
Provider(ProviderFailure),
#[error("connections is misconfigured: {0}")]
Config(String),
#[error(transparent)]
Db(#[from] cratefield_core::DbError),
}
impl ConnectionError {
#[must_use]
pub fn slug(&self) -> &'static str {
handlers::problem_for(self).slug
}
#[must_use]
pub fn problem(&self) -> Problem {
handlers::problem_for(self)
}
#[must_use]
pub fn provider_failure(&self) -> Option<&ProviderFailure> {
match self {
Self::Provider(failure) => Some(failure),
_ => None,
}
}
}
pub struct Connections {
settings: Settings,
ctx: Arc<OnceLock<Arc<ModuleContext>>>,
}
impl Default for Connections {
fn default() -> Self {
Self::builder().build()
}
}
impl Connections {
#[must_use]
pub fn builder() -> ConnectionsBuilder {
ConnectionsBuilder::new()
}
#[must_use]
pub fn api(&self) -> ConnectionsApi {
ConnectionsApi {
ctx: Arc::clone(&self.ctx),
settings: self.settings.clone(),
}
}
}
pub struct ConnectionsBuilder {
settings: Settings,
}
impl ConnectionsBuilder {
fn new() -> Self {
Self {
settings: Settings {
providers: Vec::new(),
allowed_origins: Vec::new(),
refresh_lead_secs: service::DEFAULT_REFRESH_LEAD_SECS,
},
}
}
#[must_use]
pub fn provider(mut self, provider: Provider) -> Self {
self.settings
.providers
.retain(|existing| existing.key != provider.key);
self.settings.providers.push(provider);
self
}
#[must_use]
pub fn providers<I>(mut self, providers: I) -> Self
where
I: IntoIterator<Item = Provider>,
{
for provider in providers {
self = self.provider(provider);
}
self
}
#[must_use]
pub fn allowed_origin(mut self, origin: impl AsRef<str>) -> Self {
let raw = origin.as_ref();
let normalized = origin_of(raw).unwrap_or_else(|| {
panic!("connections: `{raw}` is not an absolute http(s) origin, e.g. https://app.example.com")
});
if !self.settings.allowed_origins.contains(&normalized) {
self.settings.allowed_origins.push(normalized);
}
self
}
#[must_use]
pub fn refresh_lead(mut self, lead: Duration) -> Self {
self.settings.refresh_lead_secs = lead.whole_seconds().max(0);
self
}
#[must_use]
pub fn build(self) -> Connections {
Connections {
settings: self.settings,
ctx: Arc::new(OnceLock::new()),
}
}
}
#[derive(Clone)]
pub struct ConnectionsApi {
ctx: Arc<OnceLock<Arc<ModuleContext>>>,
settings: Settings,
}
impl ConnectionsApi {
fn ctx(&self) -> Result<Arc<ModuleContext>, ConnectionError> {
self.ctx.get().cloned().ok_or_else(|| {
ConnectionError::Config(
"the connections module has not been mounted yet; build the harness first"
.to_owned(),
)
})
}
pub async fn start(
&self,
subject: &str,
provider: &str,
return_to: &str,
) -> Result<AuthorizeUrl, ConnectionError> {
service::start(
self.ctx()?.as_ref(),
&self.settings,
subject,
provider,
return_to,
)
.await
}
pub async fn complete(&self, state: &str, code: &str) -> Result<Connection, ConnectionError> {
service::complete(self.ctx()?.as_ref(), &self.settings, state, code).await
}
pub async fn list(&self, subject: &str) -> Result<Vec<Connection>, ConnectionError> {
service::list(self.ctx()?.as_ref(), subject).await
}
pub async fn get(&self, connection_id: &str) -> Result<Connection, ConnectionError> {
service::get(self.ctx()?.as_ref(), connection_id).await
}
pub async fn access_token(&self, connection_id: &str) -> Result<AccessToken, ConnectionError> {
service::access_token(self.ctx()?.as_ref(), &self.settings, connection_id).await
}
pub async fn revoke(&self, connection_id: &str) -> Result<(), ConnectionError> {
service::revoke(self.ctx()?.as_ref(), &self.settings, connection_id).await
}
pub async fn mark_needs_reconnect(
&self,
connection_id: &str,
reason: &str,
) -> Result<(), ConnectionError> {
service::mark_needs_reconnect(self.ctx()?.as_ref(), connection_id, reason).await
}
pub async fn set_account(
&self,
connection_id: &str,
external_account_id: Option<&str>,
display_name: Option<&str>,
) -> Result<(), ConnectionError> {
service::set_account(
self.ctx()?.as_ref(),
connection_id,
external_account_id,
display_name,
)
.await
}
}
#[async_trait]
impl Module for Connections {
fn name(&self) -> &'static str {
"connections"
}
fn version(&self) -> &'static str {
env!("CARGO_PKG_VERSION")
}
fn requires(&self) -> &'static [Port] {
&[Port::Db, Port::HttpClient]
}
fn optional(&self) -> &'static [Port] {
&[Port::Clock, Port::IdGen, Port::Defer]
}
fn tables(&self) -> &'static [&'static str] {
&["connection", "connection_state"]
}
fn personal_data(&self) -> &'static [PersonalDataSet] {
static SETS: OnceLock<Vec<PersonalDataSet>> = OnceLock::new();
SETS.get_or_init(|| {
vec![
PersonalDataSet {
table: "connection",
subject: "subject",
kind: DataKind::Identifier,
disposition: Disposition::Erase,
description: "A third-party account you connected: the provider, the \
account id and name it gave us, the scopes, and the live status. The \
access and refresh tokens are held only as ciphertext and are never \
copied into an export or a response.",
redacted: &["access_token_sealed", "refresh_token_sealed"],
subject_via: None,
},
PersonalDataSet {
table: "connection_state",
subject: "subject",
kind: DataKind::Identifier,
disposition: Disposition::Erase,
description: "A connection attempt still in flight: where the browser was \
sent and when it lapses. The anti-forgery state is held only as a \
SHA-256 hash and the PKCE verifier only as ciphertext. Once spent, or \
ten minutes old, the row is deleted.",
redacted: &["state_hash", "verifier_sealed"],
subject_via: None,
},
]
})
.as_slice()
}
fn emits(&self) -> &'static [&'static str] {
&[
handlers::EVENT_CONNECTED,
handlers::EVENT_REFRESHED,
handlers::EVENT_REVOKED,
handlers::EVENT_NEEDS_RECONNECT,
]
}
fn migrations(&self) -> Migrations {
const MIGRATIONS: [SqlMigration; 1] = [MIGRATION_INIT];
const _: () = cratefield_core::assert_migration_set(&MIGRATIONS);
Migrations::sqlite(&MIGRATIONS)
}
fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
handlers::validate(&self.settings, cfg).into_result()
}
fn self_check(&self) -> Vec<String> {
handlers::self_check(&self.settings)
}
fn router(&self, ctx: ModuleContext) -> axum::Router {
let ctx = Arc::new(ctx);
let _ = self.ctx.set(Arc::clone(&ctx));
handlers::router(ctx, self.settings.clone())
}
fn surface(&self) -> Surface {
Surface::new().action(
Action::get("callback", "/callback/{provider}")
.policy(RoutePolicy::Open)
.outcome(Outcome::Redirect),
)
}
fn scheduled<'a>(
&'a self,
ctx: &'a ModuleContext,
cron: &'a str,
) -> BoxFuture<'a, Result<(), AnyError>> {
Box::pin(async move {
if let Err(error) = service::maintain(ctx, &self.settings).await {
let failure = error.provider_failure();
tracing::warn!(
cron,
slug = error.slug(),
status = ?failure.and_then(|failure| failure.status),
code = failure.map_or("-", |failure| failure.code.as_str()),
"the scheduled connections pass did not complete"
);
}
Ok(())
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_query_pair_goes_before_the_fragment() {
assert_eq!(
append_query("https://app.example.com/settings#tab", "connection", "c-1"),
"https://app.example.com/settings?connection=c-1#tab"
);
assert_eq!(
append_query("https://app.example.com/settings?a=1#tab", "b", "2"),
"https://app.example.com/settings?a=1&b=2#tab"
);
assert_eq!(
append_query("https://app.example.com/s", "b", "2"),
"https://app.example.com/s?b=2"
);
}
#[test]
fn only_a_short_token_survives_sanitising() {
assert_eq!(safe_token("access_denied"), "access_denied");
assert_eq!(safe_token("invalid_grant"), "invalid_grant");
assert_eq!(safe_token("line\nbreak"), "other");
assert_eq!(safe_token(""), "other");
assert_eq!(safe_token(&"x".repeat(65)), "other");
}
#[test]
#[should_panic(expected = "is not an absolute http(s) origin")]
fn a_bad_allowed_origin_panics() {
let _ = Connections::builder().allowed_origin("app.example.com");
}
}