use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::Context as _;
use knot_types::AccountDid;
use serde::Deserialize;
use sha2::{Digest, Sha256};
use crate::cob_write::WriteError;
use crate::config::ValidatedConfig;
use crate::datastore::Durability;
use crate::datastore::record::RecordGuard;
use crate::datastore::write::Published;
use crate::datastore::{SealedStore, schema};
use super::crypto::{
KeyPair, b64u_encode, client_assertion, derive_gcm_key, dpop_nonce_retry, dpop_proof,
http_client, now_secs, random_hex,
};
use super::{
GRANULAR_SCOPE, StoredSession, client_id, ensure_client_key, redirect_uri, save_session_under,
};
#[derive(Debug, Deserialize)]
struct AuthMetadata {
issuer: String,
#[serde(rename = "pushed_authorization_request_endpoint")]
par_endpoint: String,
authorization_endpoint: String,
token_endpoint: String,
}
async fn discover_auth_metadata(
http: &reqwest::Client,
config: &ValidatedConfig,
owner_did: &AccountDid,
) -> anyhow::Result<(String, AuthMetadata)> {
let atproto = config.atproto(
knot_runtime::ReqwestHttp::new(config.http_limits())
.context("build identity http transport")?,
)?;
let identity = atproto
.resolve_identity(owner_did)
.await
.context("resolve owner DID to find PDS")?;
let pds_url = identity
.pds
.url()
.as_str()
.trim_end_matches('/')
.to_string();
let res_meta_url = format!("{pds_url}/.well-known/oauth-protected-resource");
let mut auth_server_url = pds_url.clone();
if let Ok(resp) = http.get(&res_meta_url).send().await
&& resp.status().is_success()
&& let Ok(v) = resp.json::<serde_json::Value>().await
&& let Some(arr) = v["authorization_servers"].as_array()
&& let Some(first) = arr.first().and_then(|s| s.as_str())
{
auth_server_url = first.trim_end_matches('/').to_string();
}
let auth_meta_url = format!("{auth_server_url}/.well-known/oauth-authorization-server");
let resp = http
.get(&auth_meta_url)
.send()
.await
.context("fetch oauth authorization server metadata")?;
anyhow::ensure!(
resp.status().is_success(),
"auth server metadata failed with status {}",
resp.status()
);
let meta: AuthMetadata = resp.json().await.context("parse auth server metadata")?;
Ok((pds_url, meta))
}
pub async fn run_interactive_auth(
config: &ValidatedConfig,
owner_did: &AccountDid,
) -> anyhow::Result<StoredSession> {
let http = http_client()?;
let client_key = ensure_client_key(config)?;
let dpop_key = KeyPair::generate();
let (pds_url, meta) = discover_auth_metadata(&http, config, owner_did).await?;
let scope = GRANULAR_SCOPE;
let code_verifier = random_hex(32);
let challenge_hash = Sha256::digest(code_verifier.as_bytes());
let code_challenge = b64u_encode(&challenge_hash);
let state = random_hex(STATE_BYTES);
let callback = Callback::expect(config, &state).await?;
let ruri = redirect_uri(config);
let cid = client_id(config);
let mut nonce: Option<String> = None;
let request_uri = loop {
let assertion = client_assertion(&client_key, config, &meta.issuer)?;
let proof = dpop_proof(
&dpop_key,
"POST",
&meta.par_endpoint,
nonce.as_deref(),
None,
)?;
let params = [
("client_id", cid.as_str()),
(
"client_assertion_type",
"urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
),
("client_assertion", &assertion),
("response_type", "code"),
("code_challenge", &code_challenge),
("code_challenge_method", "S256"),
("state", &state),
("redirect_uri", &ruri),
("scope", scope),
("login_hint", owner_did.as_str()),
];
let resp = http
.post(&meta.par_endpoint)
.header("DPoP", proof)
.form(¶ms)
.send()
.await
.context("send PAR request")?;
if dpop_nonce_retry(&resp, &mut nonce) {
continue;
}
anyhow::ensure!(
resp.status().is_success(),
"PAR failed with HTTP {}: {}",
resp.status(),
resp.text().await.unwrap_or_default()
);
let par_body: serde_json::Value = resp.json().await.context("parse PAR response")?;
let r_uri = par_body["request_uri"]
.as_str()
.context("PAR response missing request_uri")?
.to_string();
break r_uri;
};
let mut auth_url = url::Url::parse(&meta.authorization_endpoint)?;
auth_url
.query_pairs_mut()
.append_pair("client_id", &cid)
.append_pair("request_uri", &request_uri);
eprintln!(
"
============================================================
knot-static: to authorize publishing repo declarations,
open this URL in your browser:
{auth_url}
waiting for authorization callback (timeout {FLOW_SECS}s)...
============================================================
"
);
let code = tokio::select! {
res = callback.code(&state) => res?,
_ = tokio::time::sleep(Duration::from_secs(FLOW_SECS)) => {
anyhow::bail!("timed out waiting for OAuth callback");
}
};
eprintln!("authorization code received; completing token exchange...");
let mut tok_nonce = nonce;
let token_data = loop {
let assertion = client_assertion(&client_key, config, &meta.issuer)?;
let proof = dpop_proof(
&dpop_key,
"POST",
&meta.token_endpoint,
tok_nonce.as_deref(),
None,
)?;
let params = [
("grant_type", "authorization_code"),
("code", code.as_str()),
("redirect_uri", &ruri),
("code_verifier", &code_verifier),
("client_id", &cid),
(
"client_assertion_type",
"urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
),
("client_assertion", &assertion),
];
let resp = http
.post(&meta.token_endpoint)
.header("DPoP", proof)
.form(¶ms)
.send()
.await
.context("token exchange request")?;
if dpop_nonce_retry(&resp, &mut tok_nonce) {
continue;
}
anyhow::ensure!(
resp.status().is_success(),
"token exchange failed with HTTP {}: {}",
resp.status(),
resp.text().await.unwrap_or_default()
);
let v: serde_json::Value = resp.json().await.context("parse token response")?;
break v;
};
let access_token = token_data["access_token"]
.as_str()
.context("missing access_token")?
.to_string();
let refresh_token = token_data["refresh_token"]
.as_str()
.context("missing refresh_token")?
.to_string();
let granted_scope = token_data["scope"].as_str().unwrap_or(scope).to_string();
let expires_in = token_data["expires_in"].as_i64().unwrap_or(3600);
let sub = token_data["sub"].as_str().unwrap_or_default();
if !sub.is_empty() && sub != owner_did.as_str() {
anyhow::bail!("authenticated account ({sub}) does not match expected owner ({owner_did})");
}
let session = StoredSession {
owner_did: owner_did.clone(),
dpop_scalar: dpop_key.to_bytes(),
access_token,
refresh_token,
scope: granted_scope,
token_endpoint: meta.token_endpoint,
pds_url,
expires_at: now_secs() + expires_in,
dpop_nonce: tok_nonce,
};
let store = SealedStore::open(config);
let record = store.record::<schema::OauthSession>(owner_did);
save_session_under(&record.guard()?, &derive_gcm_key(config)?, &session)?;
eprintln!(
"OAuth authorization successful! Tokens saved for {owner_did}.
"
);
Ok(session)
}
const FLOW_SECS: u64 = 180;
const STATE_BYTES: usize = 16;
const MAX_CODE_BYTES: usize = 2048;
enum Callback {
/// the cgi drops the code in `pending_dir`, and that dir only exists while the flow waits.
File(PendingFlow),
/// without a cgi, nginx proxies the callback to a fixed port that one flow can hold.
Listener(tokio::net::TcpListener),
}
impl Callback {
async fn expect(config: &ValidatedConfig, state: &str) -> anyhow::Result<Self> {
if config.writes.fastcgi_socket.is_some() {
return Ok(Self::File(PendingFlow::open(&pending_dir(config), state)?));
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:8918")
.await
.context(
"bind oauth callback listener on 127.0.0.1:8918 (is another flow in progress? \
flows only run side by side with writes.fastcgi_socket set)",
)?;
Ok(Self::Listener(listener))
}
async fn code(self, state: &str) -> anyhow::Result<String> {
match self {
Self::File(flow) => flow.code().await,
Self::Listener(listener) => accept_oauth_code(listener, state).await,
}
}
}
fn pending_dir(config: &ValidatedConfig) -> PathBuf {
config.output_root.join("oauth-pending")
}
fn valid_state(state: &str) -> bool {
state.len() == STATE_BYTES * 2
&& state
.bytes()
.all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
}
/// the cgi writes `<state>.code` whole, and the flow removes both files when it ends.
struct PendingFlow {
marker: PathBuf,
code: PathBuf,
}
impl PendingFlow {
fn open(dir: &Path, state: &str) -> anyhow::Result<Self> {
use std::os::unix::fs::{DirBuilderExt as _, OpenOptionsExt as _};
anyhow::ensure!(valid_state(state), "malformed oauth state");
std::fs::DirBuilder::new()
.recursive(true)
.mode(0o700)
.create(dir)
.context("create the oauth callback directory")?;
crate::datastore::marks::expire(dir, Duration::from_secs(FLOW_SECS));
let marker = dir.join(state);
std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&marker)
.context("mark the oauth flow as waiting")?;
Ok(Self {
code: dir.join(format!("{state}.code")),
marker,
})
}
async fn code(&self) -> anyhow::Result<String> {
loop {
if let Ok(code) = std::fs::read_to_string(&self.code) {
return Ok(code);
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}
}
impl Drop for PendingFlow {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.marker);
let _ = std::fs::remove_file(&self.code);
}
}
/// the state is the flow's only secret, and an unknown one writes nothing.
pub(crate) fn deliver_code(config: &ValidatedConfig, target: &str) -> Result<(), WriteError> {
let query = target
.split('#')
.next()
.and_then(|target| target.split_once('?'))
.map_or("", |(_, query)| query);
let param = |name: &str| {
url::form_urlencoded::parse(query.as_bytes())
.find(|(key, _)| key == name)
.map(|(_, value)| value.into_owned())
};
let state = param("state")
.filter(|state| valid_state(state))
.ok_or_else(|| WriteError::invalid_request("malformed state"))?;
let code = param("code")
.filter(|code| {
(1..=MAX_CODE_BYTES).contains(&code.len()) && code.bytes().all(|b| b.is_ascii_graphic())
})
.ok_or_else(|| WriteError::invalid_request("missing or malformed code"))?;
let dir = pending_dir(config);
crate::datastore::marks::expire(&dir, Duration::from_secs(FLOW_SECS));
let waiting = std::fs::symlink_metadata(dir.join(&state)).is_ok_and(|meta| meta.is_file());
if !waiting {
return Err(WriteError::invalid_request(
"no flow is waiting for this state",
));
}
// a second code fails to link, and a killed worker leaves no file to wedge the flow.
let published = crate::datastore::write::publish_once(
&dir.join(format!("{state}.code")),
code.as_bytes(),
Durability::Rename,
crate::datastore::write::Access::Private,
)
.map_err(|error| WriteError::internal(format!("oauth callback: {error}")))?;
match published {
Published::Mine => Ok(()),
Published::Existing(_) => Err(WriteError::conflict("this flow already has its code")),
}
}
async fn accept_oauth_code(
listener: tokio::net::TcpListener,
expected_state: &str,
) -> anyhow::Result<String> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
loop {
let (mut socket, _) = listener.accept().await?;
let mut buf = [0u8; 4096];
let n = socket.read(&mut buf).await?;
let req_str = String::from_utf8_lossy(&buf[..n]);
let Some(first_line) = req_str.lines().next() else {
continue;
};
let mut parts = first_line.split_whitespace();
if parts.next() != Some("GET") {
continue;
}
let Some(path_and_query) = parts.next() else {
continue;
};
let Ok(parsed) = url::Url::parse(&format!("http://localhost{path_and_query}")) else {
continue;
};
let mut code = None;
let mut state = None;
for (k, v) in parsed.query_pairs() {
match k.as_ref() {
"code" => code = Some(v.into_owned()),
"state" => state = Some(v.into_owned()),
_ => {}
}
}
if state.as_deref() != Some(expected_state) {
let resp = "HTTP/1.1 400 Bad Request\r\nContent-Length: 13\r\nConnection: close\r\n\r\nInvalid state";
let _ = socket.write_all(resp.as_bytes()).await;
continue;
}
let Some(auth_code) = code else {
let resp = "HTTP/1.1 400 Bad Request\r\nContent-Length: 12\r\nConnection: close\r\n\r\nMissing code";
let _ = socket.write_all(resp.as_bytes()).await;
continue;
};
let body = r#"<!DOCTYPE html><html><body style="font-family:sans-serif;padding:3rem;"><h2>Authorization successful!</h2><p>You may return to your terminal.</p></body></html>"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = socket.write_all(resp.as_bytes()).await;
return Ok(auth_code);
}
}
pub(crate) async fn refresh_session(
config: &ValidatedConfig,
guard: &RecordGuard<'_, schema::OauthSession>,
session: &mut StoredSession,
) -> anyhow::Result<bool> {
let http = http_client()?;
let client_key = ensure_client_key(config)?;
let dpop_key = KeyPair::from_bytes(&session.dpop_scalar)?;
let cid = client_id(config);
let audience = url::Url::parse(&session.token_endpoint)?
.origin()
.ascii_serialization();
let mut nonce = session.dpop_nonce.clone();
let token_data = loop {
let assertion = client_assertion(&client_key, config, &audience)?;
let proof = dpop_proof(
&dpop_key,
"POST",
&session.token_endpoint,
nonce.as_deref(),
None,
)?;
let params = [
("grant_type", "refresh_token"),
("refresh_token", session.refresh_token.as_str()),
("client_id", cid.as_str()),
(
"client_assertion_type",
"urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
),
("client_assertion", &assertion),
];
let resp = http
.post(&session.token_endpoint)
.header("DPoP", proof)
.form(¶ms)
.send()
.await
.context("token refresh request")?;
if dpop_nonce_retry(&resp, &mut nonce) {
continue;
}
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
eprintln!("warning: token refresh failed ({status}): {text}");
return Ok(false);
}
let v: serde_json::Value = resp.json().await.context("parse refresh response")?;
break v;
};
if let Some(new_acc) = token_data["access_token"].as_str() {
session.access_token = new_acc.to_string();
}
if let Some(new_ref) = token_data["refresh_token"].as_str() {
session.refresh_token = new_ref.to_string();
}
let expires_in = token_data["expires_in"].as_i64().unwrap_or(3600);
session.expires_at = now_secs() + expires_in;
session.dpop_nonce = nonce;
save_session_under(guard, &derive_gcm_key(config)?, session)?;
Ok(true)
}