~ knotless Search/
↑↓ move↵ open⇧↵ all results

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

Hosts identity data names are asked only on https, at public addresses

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:core/identity-guard

Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.

crates/grain/src/auth.rs+4−2
View file
@@ -158,8 +158,10 @@ mod tests {
async fn tokens_are_verified_against_the_issuers_key() {
let (port, served) = documents();
let did = format!("did:web:127.0.0.1%3A{port}");
let ctx = Ctx::new(Env::from_vars(|k| {
(k == "GRAIN_DID_WEB_SCHEME").then(|| "http".into())
let ctx = Ctx::new(Env::from_vars(|k| match k {
"GRAIN_DID_WEB_SCHEME" => Some("http".into()),
"GRAIN_ALLOW_PRIVATE" => Some("1".into()),
_ => None,
}));
let key = P256Key::generate();
*served.lock().unwrap() = document(&did, &key.public());
crates/grain/src/env.rs+46−18
View file
@@ -35,6 +35,11 @@ pub struct Env {
pub browser: Option<String>,
pub identity_ttl: Duration,
pub timeout: Duration,
/// Whether hosts identity data names may be asked at loopback, private
/// and other addresses that are not the public internet's, and not on
/// https (`GRAIN_ALLOW_PRIVATE=1`): for tests and dev only, as a local
/// PDS is. See [`crate::guard`].
pub allow_private: bool,
}
impl Env {
@@ -92,37 +97,54 @@ impl Env {
browser: get("GRAIN_BROWSER").filter(|v| !v.is_empty()),
identity_ttl: Duration::from_secs(num("GRAIN_IDENTITY_TTL_SECS", 3600)),
timeout: Duration::from_secs(num("GRAIN_TIMEOUT_SECS", 60)),
allow_private: get("GRAIN_ALLOW_PRIVATE").as_deref() == Some("1"),
}
}
}
/// Configuration plus the one HTTP client a process shares, so every request
/// Configuration plus the HTTP clients a process shares, so every request
/// to a PDS reuses its connection.
#[derive(Clone)]
pub struct Ctx {
pub env: Env,
/// For the hosts this process is set to ask, and its own account's.
pub http: reqwest::Client,
/// For the hosts identity data names, held to [`crate::guard`]: a
/// `did:web` host, a handle's, the PDS a DID document names.
pub guarded: reqwest::Client,
}
impl Ctx {
pub fn new(env: Env) -> Ctx {
let mut http = reqwest::Client::builder();
for cert in env.ca_certs.iter().flat_map(|path| extra_roots(path)) {
http = http.add_root_certificate(cert);
let base = || {
let mut http = reqwest::Client::builder();
for cert in env.ca_certs.iter().flat_map(|path| extra_roots(path)) {
http = http.add_root_certificate(cert);
}
http.user_agent(concat!("grain/", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(10))
.timeout(env.timeout)
// Under the 5s Node servers keep an idle connection by
// default, as a PLC directory does: past it, one can close
// under the request reusing it. A PDS keeps one 90s, which
// ties with any longer wait here.
.pool_idle_timeout(Duration::from_secs(4))
.tcp_nodelay(true)
};
let built = "the HTTP client configuration is static";
let http = base().build().expect(built);
let guarded = crate::guard::client(base(), env.allow_private);
let guarded = guarded.build().expect(built);
Ctx { env, http, guarded }
}
/// Whether `url`, as identity data names it, may be asked
/// ([`crate::guard::may_ask`]): any may when private hosts are allowed.
pub fn may_ask(&self, url: &str) -> Result<(), crate::guard::Refused> {
if self.env.allow_private {
return Ok(());
}
let http = http
.user_agent(concat!("grain/", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(10))
.timeout(env.timeout)
// Under the 5s Node servers keep an idle connection by default,
// as a PLC directory does: past it, one can close under the
// request reusing it. A PDS keeps one 90s, which ties with any
// longer wait here.
.pool_idle_timeout(Duration::from_secs(4))
.tcp_nodelay(true)
.build()
.expect("the HTTP client configuration is static");
Ctx { env, http }
crate::guard::may_ask(url)
}
}
@@ -174,6 +196,9 @@ mod tests {
assert!(e.service.is_none());
assert!(e.ca_certs.is_none());
assert!(e.identifier.is_none() && e.password.is_none() && e.browser.is_none());
assert!(!e.allow_private);
let ctx = Ctx::new(e);
assert!(ctx.may_ask("http://127.0.0.1:2583").is_err());
let bare = env(&[]);
assert_eq!(bare.cache_dir, PathBuf::from("./.cache/grain"));
assert_eq!(bare.config_dir, PathBuf::from("./.config/grain"));
@@ -193,6 +218,7 @@ mod tests {
("GRAIN_SERVICE", "http://pds"),
("GRAIN_IDENTITY_TTL_SECS", "5"),
("GRAIN_TIMEOUT_SECS", "soon"),
("GRAIN_ALLOW_PRIVATE", "1"),
]);
assert_eq!(e.cache_dir, PathBuf::from("/x/grain"));
assert_eq!(e.config_dir, PathBuf::from("/y/grain"));
@@ -204,6 +230,8 @@ mod tests {
assert_eq!(e.service.as_deref(), Some("http://pds"));
assert_eq!(e.identity_ttl, Duration::from_secs(5));
assert_eq!(e.timeout, Duration::from_secs(60));
assert!(e.allow_private);
assert!(!env(&[("GRAIN_ALLOW_PRIVATE", "yes")]).allow_private);
assert_eq!(
e.handle_resolvers,
vec![
@@ -224,7 +252,7 @@ mod tests {
env(&[("GRAIN_CONFIG_DIR", "/k")]).config_dir,
PathBuf::from("/k")
);
let _ = Ctx::new(e);
assert_eq!(Ctx::new(e).may_ask("http://127.0.0.1:2583"), Ok(()));
}
/// A throwaway self-signed CA, only for reading back.
crates/grain/src/guard.rsnew+180−0
View file
@@ -0,0 +1,180 @@
//! What a host that identity data names may be asked. A DID document names
//! whatever PDS it likes, a `did:web` whatever host, a handle whatever
//! name, and anyone can make grain-serve look one up, so such a host is
//! asked only on https, at a public address, its hostname resolved to
//! public addresses only ([`PublicOnly`]), and no redirect is followed
//! ([`client`]). `GRAIN_ALLOW_PRIVATE=1` lifts this for tests and dev
//! ([`crate::env::Env::allow_private`]).
//!
//! The PLC directory, the handle resolvers and the services a process is
//! configured with are its settings, not identity data, and are asked as
//! they are set.
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
/// Why a host identity data names is not asked.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Refused(pub String);
impl std::fmt::Display for Refused {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for Refused {}
/// Whether `ip` is the public internet's: not loopback, private,
/// link-local, shared, reserved, multicast, or for documentation.
pub fn public(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => {
let [a, b, c, _] = v4.octets();
let special = v4.is_private() || v4.is_loopback() || v4.is_link_local();
let special = special || v4.is_documentation() || v4.is_multicast();
let special = special || a == 0 || a >= 240 || (a == 100 && (b & 0xc0) == 64);
!(special || (a == 198 && (b & 0xfe) == 18) || (a == 192 && b == 0 && c == 0))
}
IpAddr::V6(v6) => match v6.to_ipv4_mapped() {
Some(v4) => public(v4.into()),
// Global unicast, less the documentation prefix.
None => {
let [s, t, ..] = v6.segments();
(s & 0xe000) == 0x2000 && !(s == 0x2001 && t == 0xdb8)
}
},
}
}
/// Whether `url`, as identity data names it, may be asked: on https, and
/// not at an address that is not public. A hostname's addresses are held
/// to the same when it is resolved ([`PublicOnly`]).
pub fn may_ask(url: &str) -> Result<(), Refused> {
let parsed = reqwest::Url::parse(url).map_err(|e| Refused(format!("{url}: {e}")))?;
if parsed.scheme() != "https" {
return Err(Refused(format!("{url} is not on https")));
}
let host = parsed.host_str().unwrap_or_default();
let ip = host.trim_matches(['[', ']']).parse::<IpAddr>().ok();
if ip.is_some_and(|ip| !public(ip)) {
return Err(Refused(format!("{url} is not at a public address")));
}
Ok(())
}
/// Resolves a hostname to its public addresses only, so that a name
/// pointing inward is not asked.
pub struct PublicOnly;
impl reqwest::dns::Resolve for PublicOnly {
fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
Box::pin(async move {
let found = tokio::net::lookup_host((name.as_str(), 0)).await?;
let found: Vec<SocketAddr> = found.filter(|a| public(a.ip())).collect();
let why = format!("{} has no public address", name.as_str());
if found.is_empty() {
return Err(why.into());
}
Ok(Box::new(found.into_iter()) as reqwest::dns::Addrs)
})
}
}
/// `http`, held to this policy: no redirects, since one could lead
/// anywhere the name could not, and hostnames resolved to public addresses
/// only, unless `allow_private`.
pub fn client(http: reqwest::ClientBuilder, allow_private: bool) -> reqwest::ClientBuilder {
let http = http.redirect(reqwest::redirect::Policy::none());
if allow_private {
http
} else {
http.dns_resolver(Arc::new(PublicOnly))
}
}
#[cfg(test)]
mod tests {
use super::*;
use reqwest::dns::Resolve;
#[test]
fn only_public_addresses() {
for ip in ["8.8.8.8", "1.1.1.1", "2606:4700::1111", "::ffff:8.8.8.8"] {
assert!(public(ip.parse().unwrap()), "{ip}");
}
let inward = [
"127.0.0.1",
"10.1.2.3",
"172.16.0.1",
"192.168.1.1",
"169.254.169.254",
"100.64.0.1",
"0.0.0.0",
"255.255.255.255",
"240.0.0.1",
"198.18.0.1",
"192.0.0.8",
"192.0.2.1",
"224.0.0.1",
"::1",
"::",
"fc00::1",
"fd12::1",
"fe80::1",
"ff02::1",
"2001:db8::1",
"64:ff9b::a00:1",
"::ffff:127.0.0.1",
];
for ip in inward {
assert!(!public(ip.parse().unwrap()), "{ip}");
}
}
#[test]
fn urls_asked() {
assert_eq!(may_ask("https://pds.example"), Ok(()));
assert_eq!(may_ask("https://8.8.8.8/xrpc"), Ok(()));
let refused = |url| may_ask(url).unwrap_err().to_string();
assert_eq!(
refused("http://pds.example"),
"http://pds.example is not on https"
);
for url in [
"https://127.0.0.1:2583",
"https://[::1]",
"https://10.0.0.2",
"https://169.254.169.254/latest/meta-data",
] {
assert_eq!(refused(url), format!("{url} is not at a public address"));
}
assert!(refused("not a url").starts_with("not a url: "));
}
#[tokio::test]
async fn names_resolve_to_public_addresses_only() {
let resolve = |name: &str| PublicOnly.resolve(name.parse().unwrap());
let found: Vec<SocketAddr> = resolve("8.8.8.8").await.unwrap().collect();
assert_eq!(found, ["8.8.8.8:0".parse().unwrap()]);
for name in ["127.0.0.1", "localhost"] {
let why = resolve(name).await.err().unwrap().to_string();
assert_eq!(why, format!("{name} has no public address"));
}
}
#[tokio::test]
async fn a_guarded_client_asks_no_name_that_points_inward() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let http = client(reqwest::Client::builder(), false).build().unwrap();
let err = http
.get(format!("http://localhost:{port}/"))
.send()
.await
.unwrap_err();
let chain = format!("{:?}", anyhow::Error::from(err));
assert!(chain.contains("localhost has no public address"), "{chain}");
client(reqwest::Client::builder(), true).build().unwrap();
}
}
crates/grain/src/identity.rs+76−9
View file
@@ -1,5 +1,9 @@
//! Handle → DID → PDS, and the keys a DID document names: cached on disk
//! so a warm lookup spends no round trip on it.
//!
//! The hosts identity data names, a handle's, a `did:web`'s and the PDS a
//! DID document names, are held to [`crate::guard`]; the PLC directory and
//! the handle resolvers set are asked as they are set.
use crate::crypto::PublicKey;
use crate::env::{Ctx, HandleResolver};
@@ -278,17 +282,25 @@ pub async fn resolve_handle(ctx: &Ctx, handle: &str) -> Result<String> {
pub const LOOKUP: Duration = Duration::from_secs(3);
async fn well_known(ctx: &Ctx, scheme: &str, handle: &str) -> Result<String> {
let req = ctx
.http
.get(format!("{scheme}://{handle}/.well-known/atproto-did"))
.timeout(LOOKUP);
let url = format!("{scheme}://{handle}/.well-known/atproto-did");
ctx.may_ask(&url)?;
let req = ctx.guarded.get(url).timeout(LOOKUP);
let res = crate::telemetry::send(req, 0).await?.error_for_status()?;
Ok(res.text().await?.trim().to_string())
}
pub async fn resolve_pds(ctx: &Ctx, did: &str) -> Result<String> {
let doc = document(ctx, did).await?;
service(&doc, "#atproto_pds").ok_or_else(|| anyhow!("{did} names no #atproto_pds service"))
pds(ctx, did, &doc)
}
/// The PDS `doc`, `did`'s DID document, names, if it may be asked
/// ([`Ctx::may_ask`]); a refusal is a [`crate::guard::Refused`].
pub fn pds(ctx: &Ctx, did: &str, doc: &Value) -> Result<String> {
let pds = service(doc, "#atproto_pds");
let pds = pds.ok_or_else(|| anyhow!("{did} names no #atproto_pds service"))?;
ctx.may_ask(&pds)?;
Ok(pds)
}
/// Where `did`'s DID document is read: the PLC directory, or its
@@ -317,10 +329,12 @@ const KEPT_FOR: Duration = Duration::from_secs(300);
/// `did`'s DID document, from the PLC directory or its `did:web` host.
pub async fn document(ctx: &Ctx, did: &str) -> Result<Value> {
let url = document_url(ctx, did)?;
let mut req = ctx.http.get(&url);
if did.starts_with("did:web:") {
req = req.timeout(LOOKUP);
}
let req = if did.starts_with("did:web:") {
ctx.may_ask(&url)?;
ctx.guarded.get(&url).timeout(LOOKUP)
} else {
ctx.http.get(&url)
};
let res = crate::telemetry::send(req, 0).await?;
let res = res.error_for_status().context(format!("resolving {did}"))?;
let doc: Value = res.json().await?;
@@ -570,6 +584,7 @@ mod tests {
"GRAIN_CACHE_DIR" => Some(cache_dir.clone()),
"GRAIN_DID_WEB_SCHEME" => Some("http".into()),
"GRAIN_HANDLE_RESOLVER" => Some("well-known:http".into()),
"GRAIN_ALLOW_PRIVATE" => Some("1".into()),
_ => None,
}));
let id = resolve(&ctx, &name).await.unwrap();
@@ -607,6 +622,58 @@ mod tests {
assert_eq!(handle(&ctx, nowhere).await, None);
}
#[tokio::test]
async fn hosts_identity_data_names_are_asked_only_if_public() {
let dir = tempfile::tempdir().unwrap();
let cache_dir = dir.path().to_str().unwrap().to_string();
let ctx = |scheme: &'static str| {
let cache_dir = cache_dir.clone();
Ctx::new(Env::from_vars(move |k| match k {
"GRAIN_CACHE_DIR" => Some(cache_dir.clone()),
"GRAIN_DID_WEB_SCHEME" => Some(scheme.into()),
"GRAIN_HANDLE_RESOLVER" => Some(format!("well-known:{scheme}")),
_ => None,
}))
};
let refused = |err: anyhow::Error| {
let refused = err.downcast_ref::<crate::guard::Refused>();
refused
.map(|r| r.0.clone())
.unwrap_or_else(|| panic!("{err:#}"))
};
// A did:web host on loopback, over http or https.
let did = "did:web:127.0.0.1%3A2583";
let why = refused(document(&ctx("http"), did).await.unwrap_err());
assert_eq!(
why,
"http://127.0.0.1:2583/.well-known/did.json is not on https"
);
let why = refused(document(&ctx("https"), did).await.unwrap_err());
assert!(why.ends_with("is not at a public address"), "{why}");
// A handle's host, over http.
let why = resolve_handle(&ctx("http"), "alice.test")
.await
.unwrap_err();
assert!(
why.to_string()
.contains("alice.test/.well-known/atproto-did is not on https")
);
// A DID document's PDS.
let ctx = ctx("https");
let doc = |pds: &str| json!({"service": [{"id": "#atproto_pds", "serviceEndpoint": pds}]});
let pds = |at: &str| pds(&ctx, "did:plc:a", &doc(at));
assert_eq!(pds("https://pds.example/").unwrap(), "https://pds.example");
for inward in [
"http://pds.example",
"https://10.0.0.1",
"https://[::1]:2583",
] {
assert!(refused(pds(inward).unwrap_err()).starts_with(inward));
}
assert!(pds("").is_err());
assert!(super::pds(&ctx, "did:plc:a", &json!({})).is_err());
}
#[test]
fn service_lookup() {
let doc = json!({"service": [
crates/grain/src/lib.rs+1−0
View file
@@ -9,6 +9,7 @@ pub mod base32;
pub mod crypto;
pub mod drisl;
pub mod env;
pub mod guard;
pub mod identity;
pub mod jetstream;
pub mod keys;
crates/grain/tests/common/mod.rs+4−1
View file
@@ -162,7 +162,8 @@ pub async fn account() -> Account {
}
/// A context whose config and cache are under `dir`, resolving handles at
/// the PDS and DIDs at the harness's PLC directory, with `extra` beside.
/// the PDS and DIDs at the harness's PLC directory, asking private hosts,
/// with `extra` beside or in their place.
pub fn ctx(dir: &Path, extra: &[(&str, &str)]) -> Ctx {
let net = net();
let dir = dir.to_str().unwrap().to_string();
@@ -179,6 +180,8 @@ pub fn ctx(dir: &Path, extra: &[(&str, &str)]) -> Ctx {
"GRAIN_CACHE_DIR" => Some(format!("{dir}/cache")),
"GRAIN_HANDLE_RESOLVER" => Some(net.pds.clone()),
"GRAIN_PLC_URL" => Some(net.plc.clone()),
// The reference PDS is on loopback, over http.
"GRAIN_ALLOW_PRIVATE" => Some("1".into()),
_ => None,
}
}))
crates/grain-serve/src/backfill.rs+9−156
View file
@@ -26,13 +26,13 @@
//! each time after twice as long, until it is given up on. A host that
//! asks to be left alone a while (429) is.
//!
//! A DID document names whatever PDS it likes, so one is asked only on
//! https, at a public address, its hostname resolved to public addresses
//! only, and no redirect is followed ([`Config::private`] lifts this for
//! tests and dev).
//! A DID document names whatever PDS it likes, so one is asked only as
//! [`grain::guard`] allows: on https, at a public address, its hostname
//! resolved to public addresses only, and no redirect followed
//! (`GRAIN_ALLOW_PRIVATE=1` lifts this for tests and dev).
use crate::index::{Applied, Index, PutError, records};
use anyhow::{Context, anyhow, ensure};
use anyhow::{Context, anyhow};
use bytes::Bytes;
use grain::crypto::PublicKey;
use grain::env::Ctx;
@@ -42,7 +42,6 @@ use serde::Deserialize;
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use std::io::Read;
use std::net::{IpAddr, SocketAddr};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore, mpsc};
@@ -81,10 +80,6 @@ pub struct Config {
pub rediscover: Duration,
/// How often the queue is looked at when nothing says to.
pub poll: Duration,
/// Whether hosts at loopback, private and other addresses that are not
/// the public internet's may be asked, and PDSes not on https: for
/// tests and dev only, since a DID document names any URL it likes.
pub private: bool,
}
impl Default for Config {
@@ -106,7 +101,6 @@ impl Default for Config {
total: Duration::from_secs(600),
rediscover: Duration::from_secs(6 * 3600),
poll: Duration::from_secs(5),
private: false,
}
}
}
@@ -115,8 +109,7 @@ impl Config {
/// From the environment: none unless `GRAIN_BACKFILL` is set, and not
/// `0`. `GRAIN_BACKFILL_DIRECTORIES` and `GRAIN_BACKFILL_SKIP` are
/// comma-separated, `GRAIN_BACKFILL_MIRROR` empty for none, and
/// `GRAIN_BACKFILL_CONCURRENCY` a number. `GRAIN_BACKFILL_PRIVATE=1`
/// lets it ask private addresses, as a dev PDS is.
/// `GRAIN_BACKFILL_CONCURRENCY` a number.
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Option<Config> {
if matches!(get("GRAIN_BACKFILL").as_deref(), None | Some("" | "0")) {
return None;
@@ -132,7 +125,6 @@ impl Config {
skip: get("GRAIN_BACKFILL_SKIP").map_or(d.skip, list),
mirror: get("GRAIN_BACKFILL_MIRROR").map_or(d.mirror, |m| list(m).into_iter().next()),
concurrency: concurrency.unwrap_or(d.concurrency).max(1),
private: get("GRAIN_BACKFILL_PRIVATE").as_deref() == Some("1"),
..d
})
}
@@ -224,61 +216,6 @@ fn host_of(url: &str) -> String {
.unwrap_or_default()
}
/// Whether `ip` is the public internet's: not loopback, private,
/// link-local, shared, reserved, multicast, or for documentation.
fn public(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(v4) => {
let [a, b, c, _] = v4.octets();
let special = v4.is_private() || v4.is_loopback() || v4.is_link_local();
let special = special || v4.is_documentation() || v4.is_multicast();
let special = special || a == 0 || a >= 240 || (a == 100 && (b & 0xc0) == 64);
!(special || (a == 198 && (b & 0xfe) == 18) || (a == 192 && b == 0 && c == 0))
}
IpAddr::V6(v6) => match v6.to_ipv4_mapped() {
Some(v4) => public(v4.into()),
// Global unicast, less the documentation prefix.
None => {
let [s, t, ..] = v6.segments();
(s & 0xe000) == 0x2000 && !(s == 0x2001 && t == 0xdb8)
}
},
}
}
/// Whether `pds`, as a DID document names it, may be asked: on https, and
/// not at an address that is not public. A hostname's addresses are held
/// to the same when it is resolved ([`PublicOnly`]).
fn may_ask(pds: &str) -> anyhow::Result<()> {
let url = reqwest::Url::parse(pds)?;
ensure!(url.scheme() == "https", "its PDS, {pds}, is not on https");
let host = url.host_str().unwrap_or_default().trim_matches(['[', ']']);
let ip = host.parse::<IpAddr>().ok();
ensure!(
ip.is_none_or(public),
"its PDS, {pds}, is not a public address"
);
Ok(())
}
/// Resolves a hostname to its public addresses only, so that a name
/// pointing inward is not asked.
struct PublicOnly;
impl reqwest::dns::Resolve for PublicOnly {
fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
Box::pin(async move {
let found = tokio::net::lookup_host((name.as_str(), 0)).await?;
let found: Vec<SocketAddr> = found.filter(|a| public(a.ip())).collect();
let why = format!("{} has no public address", name.as_str());
if found.is_empty() {
return Err(why.into());
}
Ok(Box::new(found.into_iter()) as reqwest::dns::Addrs)
})
}
}
pub struct Backfill {
index: Arc<Index>,
ctx: Ctx,
@@ -291,17 +228,10 @@ pub struct Backfill {
impl Backfill {
pub fn new(index: Arc<Index>, ctx: Ctx, config: Config) -> Arc<Backfill> {
// No redirects: one could lead anywhere a DID document could not.
let http = reqwest::Client::builder()
.user_agent(concat!("grain/", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(10))
.redirect(reqwest::redirect::Policy::none());
let http = if config.private {
http
} else {
http.dns_resolver(Arc::new(PublicOnly))
};
let http = http
.connect_timeout(Duration::from_secs(10));
let http = grain::guard::client(http, ctx.env.allow_private)
.build()
.expect("the HTTP client configuration is static");
let hosts = Hosts {
@@ -404,13 +334,7 @@ impl Backfill {
async fn read_and_apply(&self, did: &str) -> Result<Option<Done>, Failure> {
let doc = identity::document(&self.ctx, did).await.map_err(retry)?;
let key = identity::signing_key(&doc).map_err(give_up)?;
let pds = identity::service(&doc, "#atproto_pds");
let pds = pds
.with_context(|| format!("{did} names no PDS"))
.map_err(give_up)?;
if !self.config.private {
may_ask(&pds).map_err(give_up)?;
}
let pds = identity::pds(&self.ctx, did, &doc).map_err(give_up)?;
let host = host_of(&pds);
if self
.config
@@ -647,74 +571,3 @@ fn apply(index: &Index, did: &str, read: Fetched, mirrored: bool) -> Result<Done
.map_err(retry)?;
Ok(done)
}
#[cfg(test)]
mod tests {
use super::*;
use reqwest::dns::Resolve;
#[test]
fn only_public_addresses() {
for ip in ["8.8.8.8", "1.1.1.1", "2606:4700::1111", "::ffff:8.8.8.8"] {
assert!(public(ip.parse().unwrap()), "{ip}");
}
let inward = [
"127.0.0.1",
"10.1.2.3",
"172.16.0.1",
"192.168.1.1",
"169.254.169.254",
"100.64.0.1",
"0.0.0.0",
"255.255.255.255",
"240.0.0.1",
"198.18.0.1",
"192.0.0.8",
"192.0.2.1",
"224.0.0.1",
"::1",
"::",
"fc00::1",
"fd12::1",
"fe80::1",
"ff02::1",
"2001:db8::1",
"64:ff9b::a00:1",
"::ffff:127.0.0.1",
];
for ip in inward {
assert!(!public(ip.parse().unwrap()), "{ip}");
}
}
#[test]
fn pdses_asked() {
assert!(may_ask("https://pds.example").is_ok());
assert!(may_ask("https://8.8.8.8").is_ok());
let refused = |pds| may_ask(pds).unwrap_err().to_string();
assert_eq!(
refused("http://pds.example"),
"its PDS, http://pds.example, is not on https"
);
for pds in [
"https://127.0.0.1:2583",
"https://[::1]",
"https://10.0.0.2",
] {
let why = format!("its PDS, {pds}, is not a public address");
assert_eq!(refused(pds), why);
}
assert!(may_ask("not a url").is_err());
}
#[tokio::test]
async fn names_resolve_to_public_addresses_only() {
let resolve = |name: &str| PublicOnly.resolve(name.parse().unwrap());
let found: Vec<SocketAddr> = resolve("8.8.8.8").await.unwrap().collect();
assert_eq!(found, ["8.8.8.8:0".parse().unwrap()]);
for name in ["127.0.0.1", "localhost"] {
let why = resolve(name).await.err().unwrap().to_string();
assert_eq!(why, format!("{name} has no public address"));
}
}
}
crates/grain-serve/src/sync.rs+41−3
View file
@@ -8,6 +8,10 @@
//! from the PDS the repo's DID document names, and put by its URI and CID,
//! so Jetstream delivering the same record later changes nothing. A record
//! the PDS no longer has is deleted from the index.
//!
//! Whoever holds a service token names the PDS asked, by their DID
//! document, so it is asked only as [`grain::guard`] allows: on https, at
//! a public address, following no redirect.
use crate::App;
use crate::index::{Index, PutError, record_uri, records};
@@ -47,6 +51,8 @@ pub enum ReadBackError {
RepoMismatch(String),
/// A record of a collection the index does not keep.
UnknownCollection(String),
/// The repo's PDS is not one grain asks ([`grain::guard`]).
Unaskable(grain::guard::Refused),
/// The PDS failed to answer for the record, other than to say it has
/// no such record.
Pds {
@@ -69,6 +75,7 @@ impl std::fmt::Display for ReadBackError {
ReadBackError::UnknownCollection(uri) => {
write!(f, "{uri} is of a collection grain does not index")
}
ReadBackError::Unaskable(why) => write!(f, "the repo's PDS is not asked: {why}"),
ReadBackError::Pds { uri, error } => write!(f, "reading {uri}: {error:#}"),
ReadBackError::Refused { uri, why } => {
write!(f, "{uri} does not fit its lexicon: {why}")
@@ -87,6 +94,7 @@ impl From<ReadBackError> for Failure {
ReadBackError::NotARecord(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"),
ReadBackError::RepoMismatch(_) => (StatusCode::BAD_REQUEST, "RepoMismatch"),
ReadBackError::UnknownCollection(_) => (StatusCode::BAD_REQUEST, "UnknownCollection"),
ReadBackError::Unaskable(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"),
ReadBackError::Pds { .. } => (StatusCode::BAD_GATEWAY, "UpstreamFailure"),
ReadBackError::Refused { .. } => (StatusCode::BAD_REQUEST, "InvalidRequest"),
ReadBackError::Index(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError"),
@@ -101,7 +109,8 @@ impl From<ReadBackError> for Failure {
/// on, so that Jetstream keeps what it writes next.
///
/// A URI outside `did`'s repo, or of a collection the index does not keep,
/// refuses the whole call before anything is read. The records are read
/// refuses the whole call before anything is read, as does a `pds` that
/// [`Ctx::may_ask`] does not let be asked. The records are read
/// at once, and each one read is applied whatever became of the others,
/// so a call that fails part way can be made again as it was. A URI named
/// twice is read once.
@@ -127,7 +136,8 @@ pub async fn read_back(
named.push((uri, collection, rkey));
}
}
let pds = Xrpc::new(ctx.http.clone(), pds);
ctx.may_ask(pds).map_err(ReadBackError::Unaskable)?;
let pds = Xrpc::new(ctx.guarded.clone(), pds);
let read = named
.iter()
.map(|&(_, collection, rkey)| get_record(&pds, did, collection, rkey));
@@ -232,13 +242,23 @@ pub async fn index_records(
}
let pds = grain::identity::resolve_pds(&app.ctx, &did)
.await
.map_err(|e| Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", format!("{e:#}")))?;
.map_err(unresolved)?;
let done = read_back(&app.ctx, index, &did, &pds, &input.uris).await?;
tracing::info!(%did, indexed = done.indexed.len(), not_found = done.not_found.len(),
"indexRecords");
Ok(Json(done))
}
/// The caller's PDS not found, as `indexRecords` answers it: the caller's
/// to fix if their DID document names one grain does not ask, else the
/// network's.
fn unresolved(e: anyhow::Error) -> Failure {
match e.downcast::<grain::guard::Refused>() {
Ok(refused) => ReadBackError::Unaskable(refused).into(),
Err(e) => Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", format!("{e:#}")),
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -254,6 +274,13 @@ mod tests {
400,
"UnknownCollection",
),
(
ReadBackError::Unaskable(grain::guard::Refused(
"http://pds is not on https".into(),
)),
400,
"InvalidRequest",
),
(
ReadBackError::Pds {
uri: uri(),
@@ -276,6 +303,17 @@ mod tests {
"InternalServerError",
),
];
let refused = grain::guard::Refused("http://pds is not on https".into());
let failure = unresolved(refused.into());
assert_eq!(
(failure.status.as_u16(), failure.error),
(400, "InvalidRequest")
);
let failure = unresolved(anyhow::anyhow!("did:plc:a names no #atproto_pds service"));
assert_eq!(
(failure.status.as_u16(), failure.error),
(502, "UpstreamFailure")
);
for (e, status, error) in cases {
let said = e.to_string();
let failure = Failure::from(e);
crates/grain-serve/tests/backfill.rs+2−7
View file
@@ -147,8 +147,6 @@ fn config(stand: &str, mirror: bool) -> Config {
total: Duration::from_secs(60),
rediscover: Duration::from_secs(3600),
poll: Duration::from_millis(50),
// The reference PDS and the stand-in are on loopback, over http.
private: true,
}
}
@@ -443,9 +441,8 @@ async fn failures_are_tried_again_or_given_up() {
// Unless private hosts are allowed, a PDS on loopback over http, as a
// DID document may name, is given up on at once and never asked.
let mut public = config(&url, false);
public.private = false;
let public = Backfill::new(index.clone(), ctx(), public);
let public = pds::ctx(dir.path(), &[("GRAIN_ALLOW_PRIVATE", "0")]);
let public = Backfill::new(index.clone(), public, config(&url, false));
index.resync(&f.did).unwrap();
let why = public.backfill(&f.did).await.unwrap_err().to_string();
assert!(why.ends_with("is not on https"), "{why}");
@@ -519,10 +516,8 @@ fn settings() {
("GRAIN_BACKFILL_MIRROR", ""),
("GRAIN_BACKFILL_SKIP", ""),
("GRAIN_BACKFILL_CONCURRENCY", "0"),
("GRAIN_BACKFILL_PRIVATE", "1"),
]))
.unwrap();
assert!(set.private);
assert_eq!(set.directories, ["https://a.example", "https://b.example"]);
assert_eq!((set.mirror, set.skip.len(), set.concurrency), (None, 0, 1));
let settings = grain_serve::Settings::from_vars(vars(&[
crates/grain-serve/tests/index_records.rs+45−1
View file
@@ -44,11 +44,16 @@ struct Caller {
/// A fresh account on the reference PDS, logged in, and grain-serve with
/// an index of its own, taking tokens for [`AUD`].
async fn caller() -> Caller {
caller_with(&[]).await
}
/// [`caller`], with grain-serve's context set with `extra` too.
async fn caller_with(extra: &[(&str, &str)]) -> Caller {
let dir = tempfile::tempdir().unwrap();
let ctx = net::ctx(dir.path(), &[]);
let index = Arc::new(Index::memory().unwrap());
let served = common::served_app(App {
ctx: ctx.clone(),
ctx: net::ctx(&dir.path().join("serve"), extra),
index: Some(index.clone()),
did: Some(AUD.into()),
})
@@ -347,3 +352,42 @@ async fn a_pds_that_does_not_answer_fails_the_read_back() {
other => panic!("{other:?}"),
}
}
/// The reference PDS is on loopback, over http, as any DID document may
/// name one: unless private hosts are allowed, it is not asked.
#[tokio::test]
async fn a_caller_whose_pds_is_not_public_is_refused() {
let c = caller_with(&[("GRAIN_ALLOW_PRIVATE", "0")]).await;
let (uri, _) = c.write("social.grain.gallery", gallery("Inward")).await;
let (status, body) = c.index_records(&[&uri]).await;
assert_eq!(
(status, body["error"].as_str()),
(400, Some("InvalidRequest")),
"{body}"
);
let why = body["message"].as_str().unwrap();
assert!(why.ends_with("is not on https"), "{why}");
assert!(c.index.get::<Gallery>(&uri).unwrap().is_none());
// Named outright, a PDS on https at loopback is not asked, nor one
// whose name resolves to it.
let dir = tempfile::tempdir().unwrap();
let guarded = net::ctx(dir.path(), &[("GRAIN_ALLOW_PRIVATE", "0")]);
let uris = std::slice::from_ref(&uri);
let read = |pds: &'static str| read_back(&guarded, &c.index, &c.me.did, pds, uris);
match read("https://127.0.0.1:2583").await {
Err(ReadBackError::Unaskable(why)) => assert_eq!(
why.to_string(),
"https://127.0.0.1:2583 is not at a public address"
),
other => panic!("{other:?}"),
}
match read("https://localhost:2583").await {
Err(ReadBackError::Pds { error, .. }) => {
let why = format!("{error:?}");
assert!(why.contains("localhost has no public address"), "{why}");
}
other => panic!("{other:?}"),
}
assert!(c.index.get::<Gallery>(&uri).unwrap().is_none());
}