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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

Pools are read and written through grain-serve

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:spaces/pools

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

crates/grain-serve/src/index/spaces.rs+78−2
View file
@@ -1,6 +1,9 @@
//! What grain keeps of permissioned spaces: `space_support`, whether each
//! PDS was last found to serve them ([`crate::spaces::support`]). Neither
//! it nor `space_invite` is rebuilt from the network.
//! PDS was last found to serve them ([`crate::spaces::support`]), and
//! `space_invite`, whom grain shared each private gallery with, for them
//! to find it by ([`crate::spaces::gallery`]): a space leaves no trace on
//! a member's own PDS. An invite grants nothing; the space's authority
//! does. Neither is rebuilt from the network.
use super::Index;
use crate::spaces::Support;
@@ -43,6 +46,51 @@ impl Index {
}
}
const INVITES: &str = "SELECT space, author_did, created_at FROM space_invite v
WHERE v.member_did = ?1 ORDER BY v.created_at DESC LIMIT ?2";
/// A private gallery shared with an account, as grain shared it.
#[derive(Debug, Clone, PartialEq)]
pub struct Invite {
pub space: String,
pub author: String,
pub created_at: String,
}
impl Index {
/// Keeps that `author` shared `space` with `member` at `at`, unless
/// it was kept before.
pub fn invite(
&self,
space: &str,
member: &str,
author: &str,
at: &str,
) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
db.execute(
"INSERT INTO space_invite (space, member_did, author_did, created_at)
VALUES (?1, ?2, ?3, ?4) ON CONFLICT (space, member_did) DO NOTHING",
params![space, member, author, at],
)?;
Ok(())
}
/// The `limit` galleries shared with `member` latest.
pub fn invites(&self, member: &str, limit: usize) -> rusqlite::Result<Vec<Invite>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare(INVITES)?;
let rows = stmt.query_map(params![member, limit as i64], |r| {
Ok(Invite {
space: r.get(0)?,
author: r.get(1)?,
created_at: r.get(2)?,
})
})?;
rows.collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -73,4 +121,32 @@ mod tests {
let (_, age) = index.space_support("https://pds.test").unwrap().unwrap();
assert_eq!(age, i64::MAX);
}
#[test]
fn invites_are_kept_once_and_read_latest_first() {
let index = Index::memory().unwrap();
let at = |n: u8| format!("2026-10-3{n}T00:00:00.000Z");
index
.invite("at://a/space/t/1", "did:plc:m", "did:plc:a", &at(0))
.unwrap();
index
.invite("at://a/space/t/2", "did:plc:m", "did:plc:a", &at(1))
.unwrap();
index
.invite("at://a/space/t/1", "did:plc:m", "did:plc:a", &at(1))
.unwrap();
index
.invite("at://a/space/t/1", "did:plc:n", "did:plc:a", &at(0))
.unwrap();
let found = index.invites("did:plc:m", 10).unwrap();
let spaces: Vec<&str> = found.iter().map(|i| i.space.as_str()).collect();
assert_eq!(spaces, ["at://a/space/t/2", "at://a/space/t/1"]);
assert_eq!(found[1].created_at, at(0), "kept once");
assert_eq!(found[1].author, "did:plc:a");
assert_eq!(index.invites("did:plc:m", 1).unwrap().len(), 1);
index.read(|db| {
super::super::plan::sized(db);
assert_eq!(super::super::plan::scans(db, INVITES), Vec::<String>::new());
});
}
}
crates/grain-serve/src/index/tests.rs+0−5
View file
@@ -455,11 +455,6 @@ fn the_queries_still_to_port_read_by_index() {
// ones of what is still to be ported, as they will read.
let queries = [
"SELECT f.did FROM follow f WHERE f.subject = ?1 ORDER BY f.did LIMIT 50",
"SELECT count(*) FROM group_submission s WHERE s.group_did = ?1
AND NOT EXISTS (SELECT 1 FROM group_item i WHERE i.did = s.group_did AND i.gallery = s.gallery)
AND NOT EXISTS (SELECT 1 FROM group_decline d WHERE d.did = s.group_did AND d.submission = s.uri)",
"SELECT uri FROM group_declaration d WHERE d.did = ?1",
"SELECT * FROM space_invite v WHERE v.member_did = ?1 ORDER BY v.created_at DESC",
"SELECT * FROM report r WHERE r.status = ?1 ORDER BY r.created_at DESC LIMIT 50",
"SELECT * FROM label l WHERE l.src = ?1 AND l.id > ?2 ORDER BY l.id LIMIT 100",
"DELETE FROM story_view WHERE created_at < ?1",
crates/grain-serve/src/spaces/blob.rsnew+105−0
View file
@@ -0,0 +1,105 @@
//! `/space-blob?space=&repo=&cid=`: a picture a record in a space
//! references, read from `repo`'s PDS with the caller's credential
//! ([`super::credential`]), as an `<img>` asks for it with the cookie, or
//! an app with a delegation token. A caller the space does not admit is
//! answered 404, as for a picture there is not.
//!
//! It is kept by no cache, as no answer of a space's is
//! ([`super::routes`]). Its type is read from its first bytes, and
//! only a picture's is said; the rest are `application/octet-stream`,
//! under a policy that runs nothing.
use super::credential;
use crate::App;
use crate::read::Params;
use crate::xrpc::Failure;
use axum::body::Body;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::Response;
use grain::space;
use grain::xrpc::XrpcError;
use std::sync::Arc;
pub const PATH: &str = "/space-blob";
/// What a service token calling for a picture names as its method.
const LXM: &str = super::gallery::GET;
/// The type of an image whose bytes start `bytes`, if it is one served.
fn image_type(bytes: &[u8]) -> Option<&'static str> {
let at = |from: usize, sig: &[u8]| bytes.get(from..from + sig.len()) == Some(sig);
if at(0, b"\xFF\xD8\xFF") {
Some("image/jpeg")
} else if at(0, b"\x89PNG\r\n\x1a\n") {
Some("image/png")
} else if at(0, b"RIFF") && at(8, b"WEBP") {
Some("image/webp")
} else if at(0, b"GIF8") {
Some("image/gif")
} else if at(4, b"ftypavif") {
Some("image/avif")
} else {
None
}
}
pub async fn space_blob(
State(app): State<Arc<App>>,
headers: HeaderMap,
Query(params): Query<Vec<(String, String)>>,
) -> Result<Response, Failure> {
let params = Params(params);
let (space, repo, cid) = (
params.required("space")?,
params.required("repo")?,
params.required("cid")?,
);
let not_found = || Failure::new(StatusCode::NOT_FOUND, "NotFound", "No such picture.");
if space::parse(space).is_none() || !repo.starts_with("did:") {
return Err(not_found());
}
let read = async {
let credential = credential(&app, &headers, LXM, space).await?;
let read = space::blob_with(&app.ctx, &credential, repo, cid).await;
read.map_err(|e| {
// No such picture there reads as no such picture.
let gone = XrpcError::of(&e).is_some_and(|x| x.error == "BlobNotFound");
let failed = super::refused(&app, &credential, e);
if gone { super::not_member() } else { failed }
})
};
// Not theirs to read reads as no such picture.
let bytes = match read.await {
Err(e) if e.error == "NotAuthorized" => return Err(not_found()),
read => read?,
};
let kind = image_type(&bytes).unwrap_or("application/octet-stream");
let res = Response::builder()
.header(header::CONTENT_TYPE, kind)
.header(
header::CONTENT_SECURITY_POLICY,
"default-src 'none'; sandbox",
)
.header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
.body(Body::from(bytes));
Ok(res.expect("a response of static parts"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn pictures_are_told_by_their_first_bytes() {
let mut webp = b"RIFF\0\0\0\0WEBPVP8 ".to_vec();
webp.push(0);
assert_eq!(image_type(b"\xFF\xD8\xFF\xE0"), Some("image/jpeg"));
assert_eq!(image_type(b"\x89PNG\r\n\x1a\n\0"), Some("image/png"));
assert_eq!(image_type(&webp), Some("image/webp"));
assert_eq!(image_type(b"GIF89a"), Some("image/gif"));
assert_eq!(image_type(b"\0\0\0\x1cftypavif"), Some("image/avif"));
assert_eq!(image_type(b"<svg"), None);
assert_eq!(image_type(b""), None);
}
}
crates/grain-serve/src/spaces/gallery.rsnew+544−0
View file
544 lines changed: not loaded until asked for.
crates/grain-serve/src/spaces/group.rsnew+662−0
View file
662 lines changed: not loaded until asked for.
crates/grain-serve/src/spaces/mod.rs+327−2
View file
@@ -1,10 +1,335 @@
//! Permissioned spaces as grain-serve answers for them: whether an
//! account's PDS serves them ([`support`]).
//! account's PDS serves them ([`support`]), private galleries
//! ([`gallery`]) and their pictures ([`blob`]), groups ([`group`]) and
//! their pools ([`pool`]).
//!
//! Nothing in a space reaches a firehose, so none of it is in the index:
//! what a space holds is read from the repos that hold it when it is
//! asked for, with [`grain::space`].
//! asked for, with [`grain::space`]. A reader reads with a credential the
//! space's authority gives only to its members ([`credential`]), got with
//! the web session of the account asking, or exchanged for the delegation
//! token an app sends in the `atproto-space-delegation` header, which its
//! reader's PDS signed for that space. Each is read with for a minute at
//! most ([`REPROVE`]), so someone taken off a space stops reading it here
//! within that; it is dropped sooner when a host refuses it, and when its
//! reader logs out.
//!
//! Who can read what: a private gallery's records and pictures are read
//! by its author and the members its author added, and by no one else,
//! since no one else is given a credential; its member list by its author
//! alone, with their own session. A group's pool, and its member list,
//! by the members its host admits. A failure says which of three it is: not
//! a member (`NotAuthorized`), a PDS with no spaces (`SpacesUnsupported`),
//! or a host failing on the way (`UpstreamFailure`).
pub mod blob;
pub mod gallery;
pub mod group;
pub mod pool;
pub mod support;
pub use support::{Support, support};
use crate::App;
use crate::xrpc::Failure;
use axum::Router;
use axum::extract::Request;
use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header};
use axum::middleware::Next;
use axum::response::Response;
use axum::routing::{get, post};
use grain::space::{self, Credential, DELEGATION_HEADER};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
/// `router` with the spaces' routes, every answer of which is for its
/// caller alone ([`unshared`]).
pub fn routes(router: Router<Arc<App>>) -> Router<Arc<App>> {
let spaces = Router::new()
.route(
&format!("/xrpc/{}", support::GET),
get(support::get_space_support),
)
.route(
&format!("/xrpc/{}", gallery::CREATE),
post(gallery::create_private_gallery),
)
.route(
&format!("/xrpc/{}", gallery::ADD_MEMBER),
post(gallery::add_member),
)
.route(
&format!("/xrpc/{}", gallery::GET),
get(gallery::get_private_gallery),
)
.route(
&format!("/xrpc/{}", gallery::LIST_MEMBERS),
get(gallery::list_members),
)
.route(
&format!("/xrpc/{}", gallery::LIST),
get(gallery::list_private_galleries),
)
.route(
&format!("/xrpc/{}", gallery::LIST_SHARED),
get(gallery::list_shared_private_galleries),
)
.route(blob::PATH, get(blob::space_blob))
.route(&format!("/xrpc/{}", group::GET), get(group::get_group))
.route(&format!("/xrpc/{}", group::LIST), get(group::list_groups))
.route(
&format!("/xrpc/{}", group::SUBMISSIONS),
get(group::get_submissions),
)
.route(&format!("/xrpc/{}", group::JOIN), post(group::join_group))
.route(&format!("/xrpc/{}", group::LEAVE), post(group::leave_group))
.route(&format!("/xrpc/{}", pool::GET), get(pool::get_gallery))
.route(&format!("/xrpc/{}", pool::LIST), get(pool::list_galleries))
.route(&format!("/xrpc/{}", pool::FEED), get(pool::list_feed))
.route(
&format!("/xrpc/{}", pool::CREATE),
post(pool::create_gallery),
)
.route(
&format!("/xrpc/{}", pool::DELETE),
post(pool::delete_gallery),
)
.route(
&format!("/xrpc/{}", pool::COMMENT),
post(pool::create_comment),
)
.route(
&format!("/xrpc/{}", pool::UNCOMMENT),
post(pool::delete_comment),
)
.route(
&format!("/xrpc/{}", pool::FAVORITE),
post(pool::put_favorite),
)
.layer(axum::middleware::from_fn(unshared));
router.merge(spaces)
}
/// The answer to `req`, for its caller alone and kept by no cache, the
/// browser's included: it may hold what only members read, or say who is
/// one. A query is marked so for [`crate::viewer::answers`], which says it
/// varies by `cookie` and `authorization`; the picture, which is no query,
/// says so itself. Each varies by a delegation token too.
async fn unshared(req: Request, next: Next) -> Response {
let query = req.method() == Method::GET && req.uri().path().starts_with("/xrpc/");
crate::viewer::unshared();
let mut res = next.run(req).await;
let set = res.headers_mut();
let none = HeaderValue::from_static("private, no-store");
set.insert(header::CACHE_CONTROL, none);
let asked = [DELEGATION_HEADER, "cookie", "authorization"];
for name in asked.iter().take(if query { 1 } else { 3 }) {
set.append(header::VARY, HeaderValue::from_static(name));
}
res
}
/// How long a reader's credential for a space is read with before its
/// authority is asked for another: how long someone taken off a space may
/// still read it here.
pub const REPROVE: Duration = Duration::from_secs(60);
/// The credentials readers read spaces with, by reader and space, each
/// with when its authority gave it, and the delegation tokens spent here,
/// by `jti`, until they would expire.
#[derive(Default)]
pub struct Spaces {
credentials: Mutex<HashMap<(String, String), (Credential, Instant)>>,
spent: Mutex<HashMap<String, u64>>,
/// Groups' listings at their hosts, by group, as last read
/// ([`group::listing`]).
listings: Mutex<HashMap<String, (Instant, Option<Value>)>>,
/// Groups a reader was refused the member list of lately, by reader
/// and group ([`group::is_member`]).
refused: Mutex<HashMap<(String, String), Instant>>,
}
impl Spaces {
fn kept(&self, reader: &str, space: &str) -> Option<Credential> {
let kept = self.credentials.lock().unwrap();
let key = (reader.to_string(), space.to_string());
let fresh = |(c, at): &&(Credential, Instant)| c.lasts() && at.elapsed() < REPROVE;
kept.get(&key).filter(fresh).map(|(c, _)| c.clone())
}
fn keep(&self, credential: &Credential) {
let mut kept = self.credentials.lock().unwrap();
kept.retain(|_, (c, at)| c.lasts() && at.elapsed() < REPROVE);
let key = (credential.reader.clone(), credential.space.clone());
kept.insert(key, (credential.clone(), Instant::now()));
}
/// Drops what `reader` reads `space` with, a host having refused it.
pub fn forget(&self, reader: &str, space: &str) {
let key = (reader.to_string(), space.to_string());
self.credentials.lock().unwrap().remove(&key);
}
/// Forgets every credential `reader` read with, and the member lists
/// they were refused, as they log out.
pub fn log_out(&self, reader: &str) {
let theirs = |(r, _): &(String, String)| r == reader;
self.credentials.lock().unwrap().retain(|k, _| !theirs(k));
self.refused.lock().unwrap().retain(|k, _| !theirs(k));
}
/// Spends the delegation token `jti`, good until `exp`: false if it
/// was spent before.
fn spend(&self, jti: &str, exp: u64) -> bool {
let mut spent = self.spent.lock().unwrap();
let now = grain::keys::now_secs();
spent.retain(|_, until| *until >= now);
spent.insert(jti.to_string(), exp).is_none()
}
}
/// A credential to read `space`, for the account asking by `lxm`: the
/// reader the delegation token in `headers` names, or the caller with its
/// web session.
pub async fn credential(
app: &App,
headers: &HeaderMap,
lxm: &str,
space: &str,
) -> Result<Credential, Failure> {
let Some(token) = headers.get(DELEGATION_HEADER) else {
let did = crate::viewer::viewer(app, headers, lxm).await?;
let did = did.ok_or_else(Failure::authentication_required)?;
return credential_for(app, &did, lxm, space).await;
};
let refused = |why: String| Failure::new(StatusCode::UNAUTHORIZED, "InvalidToken", why);
let token = token.to_str().map_err(|_| refused("not a token".into()))?;
let delegation = grain::auth::verify_delegation(&app.ctx, token).await;
let delegation = delegation.map_err(|e| refused(format!("{e:#}")))?;
if delegation.space != space {
let why = format!("a delegation for {}, not {space}", delegation.space);
return Err(refused(why));
}
let caller = crate::viewer::viewer(app, headers, lxm).await?;
if caller.is_some_and(|did| did != delegation.did) {
let why = "a delegation for another account than the caller";
return Err(refused(why.into()));
}
if !app.spaces.spend(&delegation.jti, delegation.exp) {
return Err(refused("the delegation token was spent".into()));
}
let reader = &delegation.did;
if let Some(kept) = app.spaces.kept(reader, space) {
return Ok(kept);
}
match space::exchange(&app.ctx, space, reader, token).await {
Ok(got) => {
app.spaces.keep(&got);
Ok(got)
}
Err(e) => Err(failed(reader, e)),
}
}
/// A credential for `did` to read `space`, got with its web session from
/// its PDS, which must serve spaces.
pub async fn credential_for(
app: &App,
did: &str,
lxm: &str,
space: &str,
) -> Result<Credential, Failure> {
if let Some(kept) = app.spaces.kept(did, space) {
return Ok(kept);
}
let pds = crate::login::session(app, did).await?.service;
if !support(app, &pds, false).await?.supported {
return Err(unsupported());
}
let got = crate::write::with_session(app, did, lxm, |session| async move {
space::credential(&app.ctx, &session, space).await
});
match got.await? {
Ok(got) => {
app.spaces.keep(&got);
Ok(got)
}
Err(e) => Err(failed(did, e)),
}
}
/// A read of a space with `credential` failing, as [`failed`] answers it;
/// a refusal drops the credential.
pub fn refused(app: &App, credential: &Credential, e: anyhow::Error) -> Failure {
let key = (credential.reader.clone(), credential.space.clone());
if space::denied(&e) {
app.spaces.credentials.lock().unwrap().remove(&key);
}
failed(&credential.reader, e)
}
/// Reading a space refused: not a member.
pub fn not_member() -> Failure {
let why = "Only members can see this.";
Failure::new(StatusCode::BAD_REQUEST, "NotAuthorized", why)
}
/// `did`'s PDS serving no spaces.
pub fn unsupported() -> Failure {
let why = "Private galleries aren't available for your account yet.";
Failure::new(StatusCode::BAD_REQUEST, "SpacesUnsupported", why)
}
/// A call to a space for `did` failing, as grain-serve answers it: a
/// refusal, or a PDS with no such method (an authority with no spaces,
/// whose space there cannot be), is not being a member; anything else is
/// the host's failure.
pub fn failed(did: &str, e: anyhow::Error) -> Failure {
if space::denied(&e) || space::not_served(&e) {
return not_member();
}
tracing::info!(%did, error = %format!("{e:#}"), "a space failed to answer");
upstream(format!("a space's host failed to answer: {e:#}"))
}
/// A host failing on the way, as `why` says.
pub fn upstream(why: String) -> Failure {
Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", why)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn delegations_are_spent_once_and_forgotten_once_expired() {
let spaces = Spaces::default();
let now = grain::keys::now_secs();
assert!(spaces.spend("a", now + 60));
assert!(!spaces.spend("a", now + 60));
assert!(spaces.spend("b", now - 1));
assert!(spaces.spend("b", now + 60), "an expired one is forgotten");
}
#[test]
fn a_failure_says_what_it_is() {
let said = |status, error: &str| -> anyhow::Error {
grain::xrpc::XrpcError {
status,
error: error.into(),
message: String::new(),
reset: None,
}
.into()
};
let denied = failed("did:plc:a", said(400, "SpaceNotFound"));
assert_eq!(denied, not_member());
let unserved = failed("did:plc:a", said(501, "MethodNotImplemented"));
assert_eq!(unserved, not_member());
let broke = failed("did:plc:a", anyhow::anyhow!("connection reset"));
assert_eq!(broke.status, StatusCode::BAD_GATEWAY);
assert_eq!(broke.error, "UpstreamFailure");
}
}
crates/grain-serve/src/spaces/pool.rsnew+622−0
View file
622 lines changed: not loaded until asked for.
crates/grain-serve/src/spaces/support.rs+2−5
View file
@@ -164,7 +164,7 @@ pub async fn support(app: &App, pds: &str, force: bool) -> Result<Support, Failu
Ok(found)
}
fn index(app: &App) -> Result<&crate::index::Index, Failure> {
pub(crate) fn index(app: &App) -> Result<&crate::index::Index, Failure> {
let why = "this server keeps no index";
let none = || Failure::new(StatusCode::NOT_IMPLEMENTED, "MethodNotImplemented", why);
app.index.as_deref().ok_or_else(none)
@@ -188,10 +188,7 @@ pub async fn get_space_support(
}
};
let pds = grain::identity::resolve_pds(&app.ctx, &did).await;
let pds = pds.map_err(|e| {
let why = format!("finding {did}'s PDS: {e:#}");
Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", why)
})?;
let pds = pds.map_err(|e| super::upstream(format!("finding {did}'s PDS: {e:#}")))?;
Ok(Json(support(&app, &pds, force).await?))
}
crates/grain-serve/src/lib.rs+5−5
View file
@@ -278,6 +278,7 @@ impl Settings {
web,
admins: self.admins.clone(),
cards: og::Cards::default(),
spaces: Default::default(),
})
}
}
@@ -300,6 +301,8 @@ pub struct App {
pub admins: Vec<String>,
/// The OpenGraph cards drawn lately.
pub cards: og::Cards,
/// The credentials spaces are read with lately ([`spaces::Spaces`]).
pub spaces: spaces::Spaces,
}
impl App {
@@ -315,6 +318,7 @@ impl App {
web: None,
admins: Vec::new(),
cards: og::Cards::default(),
spaces: Default::default(),
}
}
}
@@ -344,7 +348,7 @@ pub fn check_index(path: &std::path::Path) -> anyhow::Result<String> {
/// The routes the server answers.
pub fn router(app: Arc<App>) -> Router {
moderation::routes(og::routes(read::routes(Router::new())))
spaces::routes(moderation::routes(og::routes(read::routes(Router::new()))))
.route("/healthz", get(status::healthz))
.route("/_status", get(status::status))
.route("/xrpc/_health", get(health))
@@ -379,10 +383,6 @@ pub fn router(app: Arc<App>) -> Router {
&format!("/xrpc/{}", preferences::PUT),
post(preferences::put_preferences),
)
.route(
&format!("/xrpc/{}", spaces::support::GET),
get(spaces::support::get_space_support),
)
.route(
"/xrpc/social.grain.repo.createRecord",
post(write::create_record),
crates/grain-serve/src/login.rs+1−0
View file
@@ -516,6 +516,7 @@ pub async fn logout(State(app): State<Arc<App>>, headers: HeaderMap) -> Result<R
let _ = oauth::revoke(&guarded(&app), &web.client, &kept.session, o).await;
}
index(&app).end_web_session(&did)?;
app.spaces.log_out(&did);
}
// Read once the account is logged out, so none of its cookies is left.
let mut others = others(&app, &headers, None);
crates/grain-serve/src/write.rs+37−15
View file
@@ -64,7 +64,7 @@ const PASSED: &[&str] = &[
];
/// The account writing: the web's, by its cookie, from a page of the site.
async fn writer(app: &App, headers: &HeaderMap, lxm: &str) -> Result<String, Failure> {
pub async fn writer(app: &App, headers: &HeaderMap, lxm: &str) -> Result<String, Failure> {
if headers.contains_key(header::AUTHORIZATION) {
let why = "a service token writes to its own PDS, then calls \
social.grain.sync.indexRecords";
@@ -118,23 +118,45 @@ enum Input<'a> {
Bytes(&'a Bytes, &'a str),
}
/// Calls `nsid` at `did`'s PDS with its web session, refreshed and called
/// once more if the PDS refuses its access token.
/// Calls `nsid` at `did`'s PDS with its web session ([`with_session`]).
async fn call(app: &App, did: &str, nsid: &str, input: Input<'_>) -> Result<Value, Failure> {
let sent = with_session(app, did, nsid, |session| {
let pds = Xrpc::new(app.ctx.guarded.clone(), &session.service);
let input = &input;
async move {
let auth = session.auth();
match input {
Input::Json(body) => pds.procedure(nsid, Some(body), Some(&auth)).await,
Input::Bytes(bytes, mime) => {
let bytes = bytes.to_vec();
pds.procedure_bytes(nsid, bytes, mime, Some(&auth)).await
}
}
}
});
sent.await?.map_err(|e| refused(nsid, e))
}
/// Runs `call`, which calls `nsid` at `did`'s PDS, with `did`'s web
/// session: refreshed and run once more if the PDS refuses its access
/// token, the session ended if it refuses the fresh one too, and
/// [`out_of_scope`] if it refuses for want of scope. Any other failure is
/// the call's, for the caller to answer.
pub async fn with_session<T, F, Fut>(
app: &App,
did: &str,
nsid: &str,
call: F,
) -> Result<anyhow::Result<T>, Failure>
where
F: Fn(Session) -> Fut,
Fut: std::future::Future<Output = anyhow::Result<T>>,
{
let mut session = login::session(app, did).await?;
let mut again = true;
loop {
let pds = Xrpc::new(app.ctx.guarded.clone(), &session.service);
let auth = session.auth();
let sent = match &input {
Input::Json(body) => pds.procedure(nsid, Some(body), Some(&auth)).await,
Input::Bytes(bytes, mime) => {
let bytes = bytes.to_vec();
pds.procedure_bytes(nsid, bytes, mime, Some(&auth)).await
}
};
let e = match sent {
Ok(out) => return Ok(out),
let e = match call(session.clone()).await {
Ok(out) => return Ok(Ok(out)),
Err(e) => e,
};
if oauth::out_of_scope(&e) {
@@ -154,7 +176,7 @@ async fn call(app: &App, did: &str, nsid: &str, input: Input<'_>) -> Result<Valu
index(app).end_web_session(did)?;
return Err(login::ended());
}
return Err(refused(nsid, e));
return Ok(Err(e));
}
}
crates/grain-serve/tests/host/mod.rsnew+269−0
View file
@@ -0,0 +1,269 @@
//! A group on the spaces alpha, run by a stand-in group host: an account
//! on the reference PDS that declared itself a group and made its members
//! space and its pool there, whose DID document then names the stand-in as
//! its PDS. The stand-in serves `fyi.opensocial.*` as opensocial's host
//! does, from its own `/.well-known/did.json`, and passes every other
//! request through to the reference PDS, so the group's spaces answer at
//! the address its DID document gives, as a group host's do.
#![allow(dead_code)]
use crate::net;
use axum::body::{Body, Bytes};
use axum::extract::State;
use axum::http::{HeaderMap, Method, StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use grain::crypto::P256Key;
use grain::session::Session;
use grain::space::{self, Admin};
use grain::xrpc::Xrpc;
use serde_json::{Value, json};
use std::sync::{Arc, Mutex};
pub struct Group {
pub did: String,
pub handle: String,
pub password: String,
/// The group's own session, at the reference PDS.
pub session: Session,
/// The stand-in's URL and DID.
pub host: String,
pub host_did: String,
/// Whether the host lets in whoever asks, or keeps them pending.
pub open: Arc<Mutex<bool>>,
/// What the host was asked, by method, and by whom.
pub asked: Arc<Mutex<Vec<(String, String)>>>,
}
struct Standing {
pds: String,
did: String,
host_did: String,
ctx: grain::env::Ctx,
session: Session,
open: Arc<Mutex<bool>>,
asked: Arc<Mutex<Vec<(String, String)>>>,
http: reqwest::Client,
}
/// The account a service token says sent it, unchecked: the stand-in
/// trusts whoever calls it.
fn issuer(headers: &HeaderMap) -> String {
let token = headers["authorization"].to_str().unwrap();
let token = token.strip_prefix("Bearer ").unwrap();
let claims = token.split('.').nth(1).unwrap();
let claims: Value = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(claims).unwrap()).unwrap();
claims["iss"].as_str().unwrap().to_string()
}
async fn well_known(State(s): State<Arc<Standing>>) -> Json<Value> {
Json(json!({ "id": s.host_did }))
}
async fn list_groups(State(s): State<Arc<Standing>>) -> Json<Value> {
Json(json!({ "groups": [{
"did": s.did,
"displayName": "Film club",
"description": "We shoot film.",
"url": "https://film.example",
"joinPolicy": "fyi.opensocial.defs#open",
"avatar": "bafkreiavatarcidbafkreiavatarcid",
"rules": [{ "rkey": "1", "title": "Be kind", "text": "Always." }, { "title": "untitled" }],
}]}))
}
async fn request_join(State(s): State<Arc<Standing>>, headers: HeaderMap) -> Json<Value> {
let who = issuer(&headers);
s.asked
.lock()
.unwrap()
.push(("requestJoin".into(), who.clone()));
if !*s.open.lock().unwrap() {
return Json(json!({ "status": "pending" }));
}
let admin = Admin {
ctx: &s.ctx,
session: &s.session,
};
admin
.put_member(&space::members(&s.did), &who, false)
.await
.unwrap();
admin
.put_member(&space::pool(&s.did), &who, true)
.await
.unwrap();
Json(json!({ "status": "admitted" }))
}
async fn leave_group(State(s): State<Arc<Standing>>, headers: HeaderMap) -> Response {
let who = issuer(&headers);
let left = ("leaveGroup".to_string(), who.clone());
let again = {
let mut asked = s.asked.lock().unwrap();
let again = asked.contains(&left);
asked.push(left);
again
};
if again {
let said = json!({ "error": "NotAMember", "message": "already left" });
return (StatusCode::BAD_REQUEST, Json(said)).into_response();
}
let pds = Xrpc::new(s.ctx.http.clone(), &s.session.service);
for space in [space::members(&s.did), space::pool(&s.did)] {
let body = json!({ "space": space, "did": who });
let removed: anyhow::Result<Value> = pds
.procedure(
"com.atproto.memberspace.removeMember",
Some(&body),
Some(&s.session.auth()),
)
.await;
removed.unwrap();
}
Json(json!({})).into_response()
}
/// Anything else, as the reference PDS answers it.
async fn through(
State(s): State<Arc<Standing>>,
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
) -> Response {
let path = uri.path_and_query().map(|p| p.as_str()).unwrap_or("/");
let mut req = s
.http
.request(method, format!("{}{path}", s.pds))
.body(body);
for (k, v) in &headers {
if k != "host" && k != "content-length" {
req = req.header(k, v);
}
}
let res = req.send().await.unwrap();
let mut out = Response::builder().status(res.status());
for (k, v) in res.headers() {
if k != "content-length" && k != "transfer-encoding" {
out = out.header(k, v);
}
}
out.body(Body::from(res.bytes().await.unwrap())).unwrap()
}
/// Signs the PLC operation `op` with `key`.
fn signed(mut op: Value, key: &P256Key) -> Value {
op.as_object_mut().unwrap().remove("sig");
let cbor = ciborium::Value::serialized(&op).unwrap();
let sig = key.sign(&grain::drisl::encode(&cbor));
op["sig"] = URL_SAFE_NO_PAD.encode(sig).into();
op
}
impl Group {
/// A new group, letting in whoever asks.
pub async fn new(site_ctx: &grain::env::Ctx, dir: &std::path::Path) -> Group {
let net = net::net();
let key = P256Key::generate();
let handle = net::fresh_handle();
let password = format!("{handle}-password");
let made: Value = reqwest::Client::new()
.post(format!("{}/xrpc/com.atproto.server.createAccount", net.pds))
.json(&json!({
"email": format!("{handle}@test.com"),
"handle": handle,
"password": password,
"recoveryKey": key.public().to_did_key(),
}))
.send()
.await
.unwrap()
.error_for_status()
.unwrap()
.json()
.await
.unwrap();
let did = made["did"].as_str().unwrap().to_string();
let ctx = net::ctx(dir, &[]);
let session = grain::session::create(&ctx, &handle, &password)
.await
.unwrap();
let admin = Admin {
ctx: &ctx,
session: &session,
};
admin.create(space::MEMBERS, "self").await.unwrap();
admin.create(space::POOL, "self").await.unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let host = format!("http://127.0.0.1:{port}");
let host_did = format!("did:web:groups{port}.test");
let open = Arc::new(Mutex::new(true));
let asked = Arc::new(Mutex::new(Vec::new()));
let standing = Arc::new(Standing {
pds: net.pds.trim_end_matches('/').to_string(),
did: did.clone(),
host_did: host_did.clone(),
ctx: ctx.clone(),
session: session.clone(),
open: open.clone(),
asked: asked.clone(),
http: reqwest::Client::new(),
});
let router = Router::new()
.route("/.well-known/did.json", get(well_known))
.route("/xrpc/fyi.opensocial.listGroups", get(list_groups))
.route("/xrpc/fyi.opensocial.requestJoin", post(request_join))
.route("/xrpc/fyi.opensocial.leaveGroup", post(leave_group))
.fallback(through)
.with_state(standing);
tokio::spawn(async move { axum::serve(listener, router).await });
// The DID document names the stand-in as the group's PDS, by the
// recovery key the account was made with.
let http = reqwest::Client::new();
let audit: Value = http
.get(format!("{}/{did}/log/audit", net.plc))
.send()
.await
.unwrap()
.json()
.await
.unwrap();
let last = audit.as_array().unwrap().last().unwrap();
let mut op = last["operation"].clone();
op["services"]["atproto_pds"]["endpoint"] = host.clone().into();
op["prev"] = last["cid"].clone();
let res = http
.post(format!("{}/{did}", net.plc))
.json(&signed(op, &key))
.send()
.await
.unwrap();
assert!(res.status().is_success(), "{}", res.text().await.unwrap());
grain::identity::forget(site_ctx, &did);
Group {
did,
handle,
password,
session,
host,
host_did,
open,
asked,
}
}
/// A service token of the group's, for `aud` and `lxm`: to call grain
/// as the group.
pub async fn token(&self, ctx: &grain::env::Ctx, aud: &str, lxm: &str) -> String {
grain::auth::service_token(ctx, &self.session, aud, lxm)
.await
.unwrap()
}
}
crates/grain-serve/tests/web/mod.rs+2−0
View file
@@ -53,6 +53,7 @@ pub async fn site_with(index: Index, sign_up_pds: &str, until: Option<u64>) -> S
web: Some(Arc::new(web)),
admins: Vec::new(),
cards: Default::default(),
spaces: Default::default(),
});
let served = common::served_on(listener, shared(&app));
let http = reqwest::Client::builder()
@@ -82,6 +83,7 @@ pub fn shared(app: &Arc<App>) -> App {
web: app.web.clone(),
admins: Vec::new(),
cards: Default::default(),
spaces: Default::default(),
}
}
crates/grain-serve/tests/index_records.rs+1−0
View file
crates/grain-serve/tests/notifications.rs+1−0
View file
crates/grain-serve/tests/push.rs+1−0
View file
crates/grain-serve/tests/read.rs+3−0
View file
crates/grain-serve/tests/spaces.rs+1087−3
View file
1090 lines changed: not loaded until asked for.
crates/grain-serve/tests/write.rs+38−0
View file
crates/grain-serve/Cargo.toml+1−0
View file
Cargo.lock+1−0
View file