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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

Groups start on their host, apps read label definitions, serve hands on pages

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:batch/three

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

crates/grain/src/moderation.rs+1−0
View file
@@ -29,6 +29,7 @@ pub const APPLY_LABEL: &str = "social.grain.admin.applyLabel";
pub const NEGATE_LABEL: &str = "social.grain.admin.negateLabel";
pub const LIST_DEFINITIONS: &str = "social.grain.admin.listLabelDefinitions";
pub const GET_SUBJECT: &str = "social.grain.admin.getReviewSubject";
pub const DESCRIBE_LABELS: &str = "social.grain.moderation.describeLabels";
/// The methods the command line and MCP call with a service token, which
/// an OAuth login asks to be let call ([`crate::oauth::loopback_scope`]).
crates/grain/src/oauth.rs+92−0
View file
@@ -767,6 +767,98 @@ pub async fn refresh(
})
}
/// A session for an account a server makes for `client` outside the
/// browser's way, such as a group a group host founds: `body` is posted
/// as JSON to `url` from `client`, its assertion addressed to `url`'s
/// origin, with a DPoP proof of a key made for the session and
/// `authorization` beside it, the service token of whoever asked for the
/// account. The answer is an OAuth token response, as its `session` or as
/// the whole of it, for its `sub`, with `scope` what `body` asked for.
/// It is taken only if `url` is on the account's own authorization
/// server: else any server could hand over any account. A refusal is
/// [`Refused`], with the error and message the server said.
pub async fn provisioned(
ctx: &Ctx,
client: &Client,
url: &str,
body: &Value,
authorization: &str,
) -> Result<Session> {
let key = P256Key::generate();
let at = origin(url);
// Each try is a new assertion: a server refuses one it has seen.
let attempt = |resend| {
let mut sent = body.clone();
for (name, value) in client.fields(&at) {
sent[name] = value.into();
}
let req = ctx
.http
.post(url)
.header("authorization", authorization)
.header("dpop", proof(&key, "POST", url, None))
.json(&sent);
async move {
let res = crate::telemetry::send(req, resend).await?;
keep_nonce(url, &res);
let ok = res.status().is_success();
let body = res.bytes().await?;
anyhow::Ok((ok, serde_json::from_slice(&body).unwrap_or(Value::Null)))
}
};
let mut answer = attempt(0).await?;
if answer.1["error"] == "use_dpop_nonce" {
answer = attempt(1).await?;
}
let (ok, value) = answer;
if !ok {
let said = |name: &str| value[name].as_str().map(str::to_string);
return Err(Refused {
error: said("error").unwrap_or_else(|| "refused".into()),
description: said("message")
.or_else(|| said("error_description"))
.unwrap_or_default(),
}
.into());
}
#[derive(Deserialize)]
struct Handed {
access_token: String,
#[serde(default)]
refresh_token: String,
expires_in: u64,
sub: String,
}
let handed = match value.get("session") {
Some(session) if session.is_object() => session.clone(),
_ => value,
};
let handed: Handed =
serde_json::from_value(handed).with_context(|| format!("{url} handed over no session"))?;
let id = identity::resolve(ctx, &handed.sub).await?;
let server = server(ctx, &id.pds).await?;
if origin(&server.issuer) != at {
bail!("{at} is not where {} logs in", handed.sub);
}
Ok(Session {
did: handed.sub,
handle: String::new(),
service: id.pds,
access_jwt: handed.access_token,
refresh_jwt: handed.refresh_token,
oauth: Some(OAuth {
key: STANDARD.encode(key.to_bytes()),
issuer: server.issuer,
token_endpoint: server.token_endpoint,
revocation_endpoint: server.revocation_endpoint,
client_id: client.key.as_ref().map(|_| client.id.clone()),
loopback_id: client.key.is_none().then(|| client.id.clone()),
scope: body["scope"].as_str().map(str::to_string),
expires_at: keys::now_secs() + handed.expires_in,
}),
})
}
/// Revokes an OAuth session's tokens, granted to `client`, as far as its
/// server lets it.
pub async fn revoke(ctx: &Ctx, client: &Client, session: &Session, oauth: &OAuth) -> Result<()> {
crates/grain-import/src/tests.rs+24−4
View file
@@ -142,8 +142,8 @@ INSERT INTO _story_views VALUES
('did:plc:alice', 'at://did:plc:bob/social.grain.story/3l3qo2vuowo2a', '2025-01-01T00:00:00.000Z'),
('did:plc:carol', 'at://did:plc:bob/social.grain.story/3l3qo2vuowo2a', '2025-01-02T00:00:00.000Z');
INSERT INTO _space_invites VALUES
('ats://did:plc:alice/social.grain.space/one', 'did:plc:bob', 'did:plc:alice', '2025-01-01T00:00:00.000Z'),
('ats://did:plc:alice/social.grain.space/one', 'did:plc:carol', 'did:plc:alice', '2025-01-02T00:00:00.000Z');
('at://did:plc:alice/space/social.grain.privateGallery/one', 'did:plc:bob', 'did:plc:alice', '2025-01-01T00:00:00.000Z'),
('at://did:plc:alice/space/social.grain.privateGallery/one', 'did:plc:carol', 'did:plc:alice', '2025-01-02T00:00:00.000Z');
INSERT INTO _space_support VALUES
('https://pds.example', 1, '[]', '2025-01-01T00:00:00.000Z'),
('https://old.example', 0, '["com.atproto.space.createSpace"]', '2025-01-02T00:00:00.000Z');
@@ -438,8 +438,8 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
"SELECT space, member_did, author_did, created_at FROM space_invite ORDER BY member_did"
),
[
"ats://did:plc:alice/social.grain.space/one|did:plc:bob|did:plc:alice|2025-01-01T00:00:00.000Z",
"ats://did:plc:alice/social.grain.space/one|did:plc:carol|did:plc:alice|2025-01-02T00:00:00.000Z",
"at://did:plc:alice/space/social.grain.privateGallery/one|did:plc:bob|did:plc:alice|2025-01-01T00:00:00.000Z",
"at://did:plc:alice/space/social.grain.privateGallery/one|did:plc:carol|did:plc:alice|2025-01-02T00:00:00.000Z",
]
);
assert_eq!(
@@ -533,6 +533,26 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
.map(|l| l.neg)
.collect();
assert_eq!(negated, [false, true]);
// A private gallery shared with bob is among the ones shared with
// him, as grain-serve lists them, named as a private gallery's space.
let shared = index.invites("did:plc:bob", 10).unwrap();
let shared: Vec<(&str, &str)> = shared
.iter()
.map(|i| (i.space.as_str(), i.author.as_str()))
.collect();
let space = "at://did:plc:alice/space/social.grain.privateGallery/one";
assert_eq!(shared, [(space, "did:plc:alice")]);
let named = grain::space::of_kind(space, grain::space::PRIVATE_GALLERY).unwrap();
assert_eq!(named.authority, "did:plc:alice");
// What a PDS was found to serve is what grain-serve asks before it
// asks the PDS again.
let (found, _) = index.space_support("https://pds.example").unwrap().unwrap();
assert!(found.supported && found.missing.is_empty());
let (found, _) = index.space_support("https://old.example").unwrap().unwrap();
assert_eq!(
(found.supported, found.missing),
(false, vec!["com.atproto.space.createSpace".to_string()])
);
let label = index.label(&grain_serve::index::Label {
src: "did:web:grain.social".into(),
uri: "did:plc:erin".into(),
crates/grain-serve/src/spaces/group.rs+204−3
View file
@@ -1,5 +1,5 @@
//! Groups: `social.grain.group.getGroup`, `listGroups`, `getSubmissions`,
//! `joinGroup` and `leaveGroup`.
//! `joinGroup`, `leaveGroup`, `createGroup` and `getGroupHost`.
//!
//! A group is an account that declared itself one
//! (`fyi.opensocial.declaration`, in the index), run by a group host: the
@@ -20,10 +20,16 @@
//! one the host never admitted. Joining and leaving are the host's to
//! decide, asked as the viewer by service auth their PDS signs
//! ([`call_host`]).
//!
//! Starting one takes a group host that has agreed to found groups for
//! grain, `--group-host` ([`starter`]): it founds the group with the
//! caller as its founder and hands grain a session as it
//! ([`create_group`]).
use super::gallery::parse;
use super::{credential_for, support};
use crate::App;
use crate::login;
use crate::read::{Params, Reader, list};
use crate::write::writer;
use crate::xrpc::Failure;
@@ -45,6 +51,21 @@ pub const LIST: &str = "social.grain.group.listGroups";
pub const SUBMISSIONS: &str = "social.grain.group.getSubmissions";
pub const JOIN: &str = "social.grain.group.joinGroup";
pub const LEAVE: &str = "social.grain.group.leaveGroup";
pub const CREATE: &str = "social.grain.group.createGroup";
pub const HOST: &str = "social.grain.group.getGroupHost";
/// What a group host is asked to found a group by.
const PROVISION: &str = "fyi.opensocial.provisionGroup";
/// What grain asks for of a group it starts, to speak for it on grain:
/// its profile, and its answers to what is offered to its pool.
const GROUP_SCOPE: &str = "atproto repo:social.grain.actor.profile \
repo:social.grain.group.item repo:social.grain.group.decline";
/// How long a group started waits for its declaration to reach the
/// index, and how often it looks.
const DECLARED_WITHIN: Duration = Duration::from_secs(5);
const DECLARED_EVERY: Duration = Duration::from_millis(200);
const ACCEPTANCE: &str = "fyi.opensocial.acceptance";
@@ -392,8 +413,6 @@ pub(super) async fn views(
Ok(out)
}
/// The group `actor`, a DID or handle, names: one the index has a
/// declaration of.
/// Refuses `group` unless it is a group's DID, before its host is looked
/// for or anything is written.
fn a_group(app: &App, group: &str) -> Result<(), Failure> {
@@ -404,6 +423,8 @@ fn a_group(app: &App, group: &str) -> Result<(), Failure> {
}
}
/// The group `actor`, a DID or handle, names: one the index has a
/// declaration of.
fn group_named(app: &App, actor: &str) -> Result<String, Failure> {
let index = support::index(app)?;
let did = index.read(|db| {
@@ -627,6 +648,168 @@ pub async fn leave_group(
Ok(Json(json!({})))
}
/// The group host groups are started on: where it is, the DID service
/// tokens name it by, and the domain a group's handle ends in, the first
/// it says it takes.
#[derive(Debug, Clone, PartialEq)]
pub struct Starter {
pub url: String,
pub did: String,
pub handle_domain: String,
}
/// The group host `--group-host` names, as it describes itself, kept
/// once it has; none for a server that starts no groups, or whose host
/// does not say.
pub async fn starter(app: &App) -> Option<Starter> {
let url = app.web.as_ref()?.group_host.clone()?;
if let Some(kept) = app.spaces.starter.lock().unwrap().clone() {
return Some(kept);
}
let described = described(app, url).await;
if let Err(e) = &described {
tracing::info!(error = %format!("{e:#}"), "a group host did not describe itself");
}
let starter = described.ok()?;
*app.spaces.starter.lock().unwrap() = Some(starter.clone());
Some(starter)
}
/// The group host at `url`, by its DID document and `describeServer`.
async fn described(app: &App, url: String) -> anyhow::Result<Starter> {
let doc = app.ctx.guarded.get(format!("{url}/.well-known/did.json"));
let doc: Value = doc.send().await?.error_for_status()?.json().await?;
let described: Value = Xrpc::new(app.ctx.guarded.clone(), &url)
.query("com.atproto.server.describeServer", &[], None)
.await?;
let did = doc["id"].as_str().context("it names itself by no DID")?;
let domain = described["availableUserDomains"][0].as_str();
let domain = domain.context("it takes no handles")?;
Ok(Starter {
did: did.to_string(),
handle_domain: domain.to_string(),
url,
})
}
/// `getGroupHost`: where groups are started, and the domain their handles
/// end in; nothing from a server that starts none.
pub async fn get_group_host(State(app): State<Arc<App>>) -> Json<Value> {
Json(match starter(&app).await {
Some(s) => json!({ "url": s.url, "handleDomain": s.handle_domain }),
None => json!({}),
})
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct CreateInput {
name: String,
display_name: String,
description: Option<String>,
}
/// Whether `name` can begin a handle at a PDS: 3 to 18 letters, digits
/// and hyphens, starting and ending with no hyphen.
fn a_name(name: &str) -> bool {
let ok = |c: char| c.is_ascii_lowercase() || c.is_ascii_digit();
(3..=18).contains(&name.len())
&& name.chars().all(|c| ok(c) || c == '-')
&& name.starts_with(ok)
&& name.ends_with(ok)
}
/// `createGroup`: starts a group on the group host, the caller its
/// founder. The host founds it by service auth the caller's PDS signs,
/// and hands grain a session as the group ([`login::provisioned`]); then
/// the caller has the host open its pool, grain writes its profile as the
/// group, and the caller's acceptance makes it theirs on grain at once.
/// It answers once the group's declaration, which the host writes, has
/// reached the index, or after a few seconds, when the group's page
/// looks for itself.
pub async fn create_group(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, Failure> {
let app: &App = &app;
let did = writer(app, &headers, CREATE).await?;
let input: CreateInput = parse(&body)?;
let (Some(starter), Some(web)) = (starter(app).await, app.web.as_deref()) else {
let why = "This Grain does not start groups";
return Err(Failure::new(StatusCode::BAD_REQUEST, "NotSupported", why));
};
let name = input.name.trim().to_lowercase();
if !a_name(&name) {
let why = "Use 3 to 18 letters, numbers and hyphens, not starting or ending with a hyphen";
return Err(Failure::new(StatusCode::BAD_REQUEST, "InvalidName", why));
}
let handle = format!("{name}{}", starter.handle_domain);
let mut profile = json!({
"$type": "social.grain.actor.profile",
"displayName": input.display_name,
"createdAt": grain::keys::now_iso(),
});
crate::read::views::put(&mut profile, "description", input.description.clone());
let profile = super::gallery::checked(app, "social.grain.actor.profile", profile)?;
let aud = &starter.did;
let token = super::gallery::as_author(app, &did, PROVISION, |session| async move {
grain::auth::service_token(&app.ctx, &session, aud, PROVISION).await
})
.await?;
let mut ask = json!({
"handle": handle,
"displayName": input.display_name,
"scope": GROUP_SCOPE,
});
crate::read::views::put(&mut ask, "description", input.description);
let url = format!("{}/xrpc/{PROVISION}", starter.url);
let bearer = format!("Bearer {token}");
let started = login::provisioned(app, web, &url, &ask, &bearer, &handle).await;
let group = started.map_err(|e| {
tracing::info!(%did, %handle, error = %format!("{e:#}"), "a group host did not start a group");
let refused = grain::oauth::Refused::of(&e);
match refused.is_some_and(|r| r.error.to_lowercase().contains("handle")) {
true => Failure::new(
StatusCode::BAD_REQUEST,
"HandleUnavailable",
"That handle is taken or not allowed. Try another one",
),
false => Failure::new(
StatusCode::BAD_REQUEST,
"CreateFailed",
"The group couldn't be started right now. Try again later",
),
}
})?;
// Which spaces a group has, and who reads them, is its host's to
// say, so it is the founder who asks for the pool.
let pool = json!({
"group": group,
"type": space::POOL,
"skey": "self",
"name": "Pool",
"readableBy": ["member"],
});
let opened = call_host(app, &did, &group, "fyi.opensocial.createSpace", &pool).await;
if let Err(e) = opened {
tracing::info!(%group, error = %e.message, "a group's pool was not opened");
let why = "The group was started, but its pool couldn't be set up. Try again later";
return Err(Failure::new(StatusCode::BAD_REQUEST, "PoolFailed", why));
}
let collection = "social.grain.actor.profile";
crate::write::put(app, &group, collection, "self", profile, [None, None]).await?;
// Taken as it comes: the group's page writes it if this does not.
let _ = accept(app, &did, &group).await;
let began = Instant::now();
while !is_group(app, &group)? && began.elapsed() < DECLARED_WITHIN {
tokio::time::sleep(DECLARED_EVERY).await;
}
Ok(Json(json!({ "did": group, "handle": handle })))
}
/// The queries the views of groups run, for their plans to be checked.
#[cfg(test)]
pub fn queries() -> Vec<String> {
@@ -645,6 +828,24 @@ pub fn queries() -> Vec<String> {
mod tests {
use super::*;
#[test]
fn a_name_begins_a_handle() {
for name in ["abc", "film-club", "a1b", "x23456789012345678"] {
assert!(a_name(name), "{name}");
}
for name in [
"ab",
"-abc",
"abc-",
"Film",
"film club",
"x234567890123456789",
"café",
] {
assert!(!a_name(name), "{name}");
}
}
#[test]
fn the_reads_of_groups_read_by_index() {
let index = crate::index::Index::memory().unwrap();
crates/grain-serve/src/spaces/mod.rs+11−0
View file
@@ -84,6 +84,14 @@ pub fn routes(router: Router<Arc<App>>) -> Router<Arc<App>> {
get(group::get_submissions),
)
.route(&format!("/xrpc/{}", group::JOIN), post(group::join_group))
.route(
&format!("/xrpc/{}", group::CREATE),
post(group::create_group),
)
.route(
&format!("/xrpc/{}", group::HOST),
get(group::get_group_host),
)
.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))
@@ -149,6 +157,9 @@ pub struct Spaces {
/// Groups a reader was refused the member list of lately, by reader
/// and group ([`group::is_member`]).
refused: Mutex<HashMap<(String, String), Instant>>,
/// The group host groups are started on, as it described itself
/// ([`group::starter`]).
starter: Mutex<Option<group::Starter>>,
}
impl Spaces {
crates/grain-serve/src/lib.rs+19−1
View file
@@ -64,6 +64,7 @@ pub const FLAGS: &[&str] = &[
"--cdn",
"--bsky-appview",
"--sign-up-pds",
"--group-host",
"--carry-cookies-until",
"--apns-key-id",
"--apns-team-id",
@@ -78,7 +79,7 @@ pub const FLAGS: &[&str] = &[
];
/// What `grain-serve` says of its flags when given one it does not take.
pub const USAGE: &str = "usage: grain-serve [--listen <addr>] [--index <file>] [--public-url <https://host>] [--jetstream <wss://host>] [--backfill <https://directory>[,...]] [--backfill-mirror <https://mirror>] [--backfill-skip <host>[,...]] [--backfill-concurrency <n>] [--cdn <https://imgproxy>] [--bsky-appview <https://appview>] [--sign-up-pds <https://pds>] [--carry-cookies-until <rfc3339 time>] [--apns-key-id <id>] [--apns-team-id <id>] [--apns-topic <bundle id>] [--apns-url <https://apns>] [--fcm-project-id <id>] [--clef-model <model>] [--clef-url <url>] [--check-index <file>] [--metrics <addr>] [--web <http://app>]
pub const USAGE: &str = "usage: grain-serve [--listen <addr>] [--index <file>] [--public-url <https://host>] [--jetstream <wss://host>] [--backfill <https://directory>[,...]] [--backfill-mirror <https://mirror>] [--backfill-skip <host>[,...]] [--backfill-concurrency <n>] [--cdn <https://imgproxy>] [--bsky-appview <https://appview>] [--sign-up-pds <https://pds>] [--group-host <https://host>] [--carry-cookies-until <rfc3339 time>] [--apns-key-id <id>] [--apns-team-id <id>] [--apns-topic <bundle id>] [--apns-url <https://apns>] [--fcm-project-id <id>] [--clef-model <model>] [--clef-url <url>] [--check-index <file>] [--metrics <addr>] [--web <http://app>]
environment: GRAIN_ADMIN_DIDS, GRAIN_CDN_KEY, GRAIN_CDN_SALT, GRAIN_APNS_KEY_FILE, GRAIN_FCM_KEY_FILE, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_AI_TOKEN, RUST_LOG";
/// What the server is set to do, from its flags ([`FLAGS`]) and, for
@@ -110,6 +111,10 @@ pub struct Settings {
/// `--sign-up-pds`: where someone making an account on the web makes
/// it. None is [`login::SIGN_UP_PDS`].
pub sign_up_pds: Option<String>,
/// `--group-host`: the group host the web starts groups on, which has
/// agreed to found them for grain
/// ([`spaces::group::create_group`]). None starts none.
pub group_host: Option<String>,
/// `GRAIN_ADMIN_DIDS`, from the environment: the DIDs of the accounts
/// that moderate, separated by commas or spaces ([`moderation`]).
pub admins: Vec<String>,
@@ -181,6 +186,7 @@ impl Settings {
cdn: [get("--cdn"), get("GRAIN_CDN_KEY"), get("GRAIN_CDN_SALT")],
bsky_appview: url("--bsky-appview"),
sign_up_pds: get("--sign-up-pds"),
group_host: url("--group-host"),
admins: get("GRAIN_ADMIN_DIDS")
.unwrap_or_default()
.split([',', ' '])
@@ -278,6 +284,7 @@ impl Settings {
let until = self.carry_cookies_until.as_deref();
let until = until.map(|t| index.epoch_secs(t)).transpose()?;
let web = login::Web::new(url, index, pds)?.carrying(index, until)?;
let web = web.starting_groups(self.group_host.as_deref());
Some(Arc::new(web))
}
_ => None,
@@ -526,6 +533,8 @@ mod tests {
"https://public.api.bsky.app/",
"--sign-up-pds",
"https://pds.test",
"--group-host",
"https://groups.test/",
"--jetstream",
jetstream::DEFAULT,
"--web",
@@ -557,6 +566,7 @@ mod tests {
assert_eq!(set.index, Some(PathBuf::from("/data/index.db")));
assert_eq!(set.public_url.as_deref(), Some("https://grain.social"));
assert_eq!(set.sign_up_pds.as_deref(), Some("https://pds.test"));
assert_eq!(set.group_host.as_deref(), Some("https://groups.test"));
// The last of a flag given twice is the one taken.
let twice = Settings::parse(&args(&["--index", "a.db", "--index", "b.db"]), none);
assert_eq!(twice.unwrap().index, Some(PathBuf::from("b.db")));
@@ -651,6 +661,14 @@ mod tests {
"https://grain.social/oauth-client-metadata.json"
);
assert_eq!(web.sign_up_pds, login::SIGN_UP_PDS);
assert_eq!(web.group_host, None);
let hosting = Settings {
group_host: Some("https://groups.test".into()),
..settings.clone()
};
let app = hosting.app(ctx()).unwrap();
let web = app.web.as_ref().unwrap();
assert_eq!(web.group_host.as_deref(), Some("https://groups.test"));
let bare = Settings::default().app(ctx()).unwrap();
assert!(bare.index.is_none() && bare.did.is_none() && bare.web.is_none());
let wrong = Settings {
crates/grain-serve/src/login.rs+29−0
View file
@@ -120,6 +120,8 @@ pub struct Web {
/// What signs the cookies, and nothing else.
cookie_key: P256Key,
pub sign_up_pds: String,
/// The group host groups are started on, if any: `--group-host`.
pub group_host: Option<String>,
logins: Mutex<HashMap<String, Pending>>,
/// A lock for each account whose session is being refreshed.
refreshing: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
@@ -154,12 +156,20 @@ impl Web {
client,
cookie_key,
sign_up_pds: sign_up_pds.trim_end_matches('/').to_string(),
group_host: None,
logins: Mutex::new(HashMap::new()),
refreshing: Mutex::new(HashMap::new()),
carryover: None,
})
}
/// This web, starting groups on the group host at `url`, if any
/// ([`crate::spaces::group::create_group`]).
pub fn starting_groups(mut self, url: Option<&str>) -> Web {
self.group_host = url.map(|u| u.trim_end_matches('/').to_string());
self
}
/// A session cookie's token for `did`, signed now.
fn token(&self, did: &str) -> String {
let claims = Claims {
@@ -465,6 +475,25 @@ async fn keep(
Ok(session.did)
}
/// Keeps the session a server hands this client for an account it made,
/// asked for at `url` with `body` and `authorization`
/// ([`oauth::provisioned`]), named by `handle`. Returns the account. Its
/// refusal is [`oauth::Refused`].
pub async fn provisioned(
app: &App,
web: &Web,
url: &str,
body: &Value,
authorization: &str,
handle: &str,
) -> anyhow::Result<String> {
let ctx = guarded(app);
let mut session = oauth::provisioned(&ctx, &web.client, url, body, authorization).await?;
session.handle = handle.to_string();
index(app).keep_web_session(&session, keys::now_micros())?;
Ok(session.did)
}
#[derive(Deserialize)]
pub struct Switch {
did: String,
crates/grain-serve/src/moderation.rs+29−4
View file
@@ -4,8 +4,10 @@
//! (`com.atproto.moderation.createReport`), with a service token from
//! their PDS or the web's cookie; the report is filed under the label its
//! reason asks for ([`grain::labels::label_for`]). Anyone reads the labels
//! grain gives out (`com.atproto.label.queryLabels`). An account's labels
//! name it by its bare DID, never `at://<did>`, as Bluesky's do.
//! grain gives out (`com.atproto.label.queryLabels`), and what each of them
//! means (`social.grain.moderation.describeLabels`), as a Bluesky labeler's
//! service record says. An account's labels name it by its bare DID, never
//! `at://<did>`, as Bluesky's do.
//!
//! The admins, the accounts `GRAIN_ADMIN_DIDS` names, each calling with a
//! service token for the method from their own PDS, and only so:
@@ -42,8 +44,8 @@ use std::sync::Arc;
use unicode_segmentation::UnicodeSegmentation;
pub use grain::moderation::{
APPLY_LABEL, CREATE_REPORT, GET_REPORT, GET_STATUS, GET_SUBJECT, LIST_DEFINITIONS,
LIST_REPORTS, NEGATE_LABEL, QUERY_LABELS, RESOLVE_REPORT, UPDATE_STATUS,
APPLY_LABEL, CREATE_REPORT, DESCRIBE_LABELS, GET_REPORT, GET_STATUS, GET_SUBJECT,
LIST_DEFINITIONS, LIST_REPORTS, NEGATE_LABEL, QUERY_LABELS, RESOLVE_REPORT, UPDATE_STATUS,
};
const REPO_REF: &str = "com.atproto.admin.defs#repoRef";
@@ -62,6 +64,7 @@ pub fn routes(router: Router<Arc<App>>) -> Router<Arc<App>> {
router
.route(&at(CREATE_REPORT), post(create_report))
.route(&at(QUERY_LABELS), get(query_labels))
.route(&at(DESCRIBE_LABELS), get(describe_labels))
.route(&at(UPDATE_STATUS), post(update_subject_status))
.route(&at(GET_STATUS), get(get_subject_status))
.route(&at(LIST_REPORTS), get(list_reports))
@@ -502,6 +505,28 @@ async fn labeled(
Ok(Json(json!({})))
}
/// `social.grain.moderation.describeLabels`: the labels this server gives
/// out and what each means, as `app.bsky.labeler.defs#labelerPolicies`
/// has them, under the DID they name it by in `src`. Anyone may ask: each
/// can stand on what anyone reads, with `queryLabels`, so a client needs
/// every one to show what it means. The values are its definitions' and
/// `!takedown`, which atproto defines.
pub async fn describe_labels(State(app): State<Arc<App>>) -> Json<Value> {
let defined = labels::DEFINITIONS.iter();
let values = std::iter::once(TAKEDOWN).chain(defined.clone().map(|d| d.identifier));
let definitions: Vec<Value> = defined.map(|d| d.json()).collect();
let mut out = json!({
"policies": {
"labelValues": values.collect::<Vec<_>>(),
"labelValueDefinitions": definitions,
},
});
if let Some(did) = &app.did {
out["did"] = did.as_str().into();
}
Json(out)
}
/// `social.grain.admin.listLabelDefinitions`: the labels grain defines.
pub async fn list_definitions(
State(app): State<Arc<App>>,
crates/grain-serve/src/write.rs+35−10
View file
@@ -319,21 +319,46 @@ pub async fn put_record(
) -> Result<Json<Value>, Failure> {
let did = writer(&app, &headers, "social.grain.repo.putRecord").await?;
let input: Put = parse(&body)?;
let value = record(&app, &input.collection, input.record)?;
let body = with_repo(
let swaps = [input.swap_record, input.swap_commit];
let (uri, cid) = put(
&app,
&did,
&input.collection,
&input.rkey,
input.record,
swaps,
)
.await?;
Ok(Json(json!({ "uri": uri, "cid": cid })))
}
/// Puts `value` at `rkey` in `collection` of `did`'s repo with its web
/// session, as `putRecord` does, swapping the record and commit `swaps`
/// name, and keeps it in the index. Returns its URI and CID.
pub async fn put(
app: &App,
did: &str,
collection: &str,
rkey: &str,
value: Value,
swaps: [Option<String>; 2],
) -> Result<(String, String), Failure> {
let value = record(app, collection, value)?;
let [swap_record, swap_commit] = swaps;
let body = with_repo(
did,
json!({
"collection": input.collection,
"rkey": input.rkey,
"collection": collection,
"rkey": rkey,
"record": value,
"swapRecord": input.swap_record,
"swapCommit": input.swap_commit,
"swapRecord": swap_record,
"swapCommit": swap_commit,
}),
);
let out = call(&app, &did, "com.atproto.repo.putRecord", Input::Json(&body)).await?;
let (uri, cid) = written(&did, &input.collection, &out)?;
keep(&app, &did, uri, cid, &value);
Ok(Json(json!({ "uri": uri, "cid": cid })))
let out = call(app, did, "com.atproto.repo.putRecord", Input::Json(&body)).await?;
let (uri, cid) = written(did, collection, &out)?;
keep(app, did, uri, cid, &value);
Ok((uri.to_string(), cid.to_string()))
}
#[derive(Deserialize)]
crates/grain-serve/tests/host/mod.rs+342−46
View file
@@ -5,6 +5,12 @@
//! 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.
//!
//! [`Starter`] is a stand-in for a group host that founds groups for
//! grain: it makes each group an account on the reference PDS whose DID
//! document names the stand-in, and, as the group's authorization server,
//! hands grain a session as it, a token of its own that it trades for the
//! group's session at the reference PDS on the way through.
#![allow(dead_code)]
use crate::net;
@@ -21,6 +27,7 @@ use grain::session::Session;
use grain::space::{self, Admin};
use grain::xrpc::Xrpc;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
pub struct Group {
@@ -134,12 +141,21 @@ async fn through(
uri: Uri,
headers: HeaderMap,
body: Bytes,
) -> Response {
forward(&s.http, &s.pds, method, uri, headers, body).await
}
/// The request, as the reference PDS at `pds` answers it.
async fn forward(
http: &reqwest::Client,
pds: &str,
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);
let mut req = http.request(method, format!("{pds}{path}")).body(body);
for (k, v) in &headers {
if k != "host" && k != "content-length" {
req = req.header(k, v);
@@ -164,6 +180,60 @@ fn signed(mut op: Value, key: &P256Key) -> Value {
op
}
/// Has `did`'s DID document name `host` as its PDS, by the recovery key
/// `key` the account was made with.
async fn move_to(did: &str, key: &P256Key, host: &str) {
let net = net::net();
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.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());
}
/// An account made on the reference PDS with a recovery key of the
/// test's, and its session there.
async fn made(dir: &std::path::Path, handle: &str, key: &P256Key) -> (String, Session) {
let net = net::net();
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 ctx = net::ctx(dir, &[]);
let session = grain::session::create(&ctx, handle, &password)
.await
.unwrap();
(made["did"].as_str().unwrap().to_string(), session)
}
impl Group {
/// A new group, letting in whoever asks.
pub async fn new(site_ctx: &grain::env::Ctx, dir: &std::path::Path) -> Group {
@@ -171,27 +241,8 @@ impl Group {
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 (did, session) = made(dir, &handle, &key).await;
let ctx = net::ctx(dir, &[]);
let session = grain::session::create(&ctx, &handle, &password)
.await
.unwrap();
let admin = Admin {
ctx: &ctx,
session: &session,
@@ -224,28 +275,7 @@ impl Group {
.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());
move_to(&did, &key, &host).await;
grain::identity::forget(site_ctx, &did);
Group {
did,
@@ -267,3 +297,269 @@ impl Group {
.unwrap()
}
}
/// What the stand-in that founds groups was asked, and keeps.
struct Founding {
pds: String,
url: String,
did: String,
/// The site's: where a group's declaration arrives, as Jetstream
/// would bring it, and whose identities are forgotten as they move.
site: grain::env::Ctx,
index: Arc<grain_serve::index::Index>,
dir: std::path::PathBuf,
asked: Arc<Mutex<Vec<(String, String)>>>,
/// What `provisionGroup` was sent last.
sent: Arc<Mutex<Value>>,
/// Each group's session at the reference PDS, by the token handed to
/// grain for it, and by its DID.
tokens: Mutex<HashMap<String, Session>>,
/// The `jti`s of the client assertions it was sent.
seen: Mutex<Vec<String>>,
}
pub struct Starter {
pub url: String,
pub did: String,
/// What the host was asked, by method, and by whom.
pub asked: Arc<Mutex<Vec<(String, String)>>>,
/// The input `provisionGroup` was sent last.
pub sent: Arc<Mutex<Value>>,
_dir: tempfile::TempDir,
}
/// The nonce the stand-in wants in a DPoP proof.
const NONCE: &str = "standin-nonce";
/// The claims of a JWT, unchecked.
fn claims(jwt: &str) -> Value {
let part = jwt.split('.').nth(1).unwrap_or_default();
let bytes = URL_SAFE_NO_PAD.decode(part).unwrap_or_default();
serde_json::from_slice(&bytes).unwrap_or(Value::Null)
}
fn refuse(error: &str) -> Response {
let said = json!({ "error": error, "message": format!("the stand-in says {error}") });
(StatusCode::BAD_REQUEST, Json(said)).into_response()
}
async fn host_did(State(s): State<Arc<Founding>>) -> Json<Value> {
Json(json!({ "id": s.did }))
}
async fn describe(State(s): State<Arc<Founding>>) -> Json<Value> {
Json(json!({ "did": s.did, "availableUserDomains": [".test"] }))
}
async fn resource(State(s): State<Arc<Founding>>) -> Json<Value> {
Json(json!({ "resource": s.url, "authorization_servers": [s.url] }))
}
async fn authorization(State(s): State<Arc<Founding>>) -> Json<Value> {
Json(json!({
"issuer": s.url,
"authorization_endpoint": format!("{}/oauth/authorize", s.url),
"token_endpoint": format!("{}/oauth/token", s.url),
"pushed_authorization_request_endpoint": format!("{}/oauth/par", s.url),
}))
}
/// Founds a group named by the input's `handle`, its founder whoever the
/// service token says sent it, as a group host does: unless the handle
/// begins `taken`, which it refuses; for one beginning `stolen` it hands
/// over the founder's own account instead, which it does not log in. A
/// client assertion it was sent before is refused, as a replay.
async fn provision(
State(s): State<Arc<Founding>>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Response {
let assertion = body["client_assertion"].as_str().unwrap_or_default();
let jti = claims(assertion)["jti"]
.as_str()
.unwrap_or_default()
.to_string();
let replayed = {
let mut seen = s.seen.lock().unwrap();
let again = jti.is_empty() || seen.contains(&jti);
seen.push(jti);
again
};
if replayed {
return refuse("invalid_client");
}
let proof = headers.get("dpop").and_then(|v| v.to_str().ok());
if claims(proof.unwrap_or_default())["nonce"] != NONCE {
let said = json!({ "error": "use_dpop_nonce", "message": "a nonce" });
let mut res = (StatusCode::BAD_REQUEST, Json(said)).into_response();
res.headers_mut()
.insert("dpop-nonce", NONCE.parse().unwrap());
return res;
}
let founder = issuer(&headers);
let asked = ("provisionGroup".to_string(), founder.clone());
s.asked.lock().unwrap().push(asked);
*s.sent.lock().unwrap() = body.clone();
let handle = body["handle"].as_str().unwrap().to_string();
if handle.starts_with("taken") {
return refuse("HandleNotAvailable");
}
let handed = |sub: &str, token: &str| {
Json(json!({ "session": {
"access_token": token,
"refresh_token": format!("{token}-refresh"),
"token_type": "DPoP",
"expires_in": 3600,
"scope": body["scope"],
"sub": sub,
}}))
.into_response()
};
if handle.starts_with("stolen") {
// The token response as the whole answer, as a host may give it.
let whole = json!({
"access_token": "standin-stolen",
"token_type": "DPoP",
"expires_in": 3600,
"sub": founder,
});
return Json(whole).into_response();
}
let key = P256Key::generate();
let (did, session) = made(&s.dir, &handle, &key).await;
let admin = Admin {
ctx: &net::ctx(&s.dir, &[]),
session: &session,
};
admin.create(space::MEMBERS, "self").await.unwrap();
admin
.put_member(&space::members(&did), &founder, false)
.await
.unwrap();
move_to(&did, &key, &s.url).await;
grain::identity::forget(&s.site, &did);
let declared = json!({
"$type": "fyi.opensocial.declaration",
"meta": format!("at://{did}/space/fyi.opensocial.meta/self"),
"createdAt": "2026-10-31T12:00:00.000Z",
});
let uri = format!("at://{did}/fyi.opensocial.declaration/self");
let cid = "bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm";
// A moment after the host answers, as Jetstream brings it.
let index = s.index.clone();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
index.put(&uri, cid, &declared).unwrap();
});
let token = format!("standin-{did}");
let mut tokens = s.tokens.lock().unwrap();
tokens.insert(token.clone(), session.clone());
tokens.insert(did.clone(), session);
drop(tokens);
handed(&did, &token)
}
/// Opens a group's space for whoever the service token says sent it,
/// letting them write there: unless the group's handle begins `nopool`.
async fn create_space(
State(s): State<Arc<Founding>>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Response {
let who = issuer(&headers);
s.asked
.lock()
.unwrap()
.push(("createSpace".into(), who.clone()));
let group = body["group"].as_str().unwrap_or_default();
let session = s.tokens.lock().unwrap().get(group).cloned();
let Some(session) = session.filter(|g| !g.handle.starts_with("nopool")) else {
return refuse("InvalidRequest");
};
let kind = body["type"].as_str().unwrap();
let admin = Admin {
ctx: &net::ctx(&s.dir, &[]),
session: &session,
};
let uri = admin.create(kind, "self").await.unwrap();
let space = format!("at://{group}/space/{kind}/self");
admin.put_member(&space, &who, true).await.unwrap();
Json(json!({ "uri": uri })).into_response()
}
/// Anything else, as the reference PDS answers it, a token the stand-in
/// handed over traded for its group's session there.
async fn founded(
State(s): State<Arc<Founding>>,
method: Method,
uri: Uri,
mut headers: HeaderMap,
body: Bytes,
) -> Response {
let said = headers.get("authorization").and_then(|v| v.to_str().ok());
let token = said.and_then(|a| a.strip_prefix("DPoP "));
let session = token.and_then(|t| s.tokens.lock().unwrap().get(t).cloned());
if let Some(session) = session {
let bearer = format!("Bearer {}", session.access_jwt);
headers.insert("authorization", bearer.parse().unwrap());
headers.remove("dpop");
}
forward(&reqwest::Client::new(), &s.pds, method, uri, headers, body).await
}
impl Starter {
/// Where a stand-in will listen, for the site to be told before it
/// starts.
pub async fn bind() -> (tokio::net::TcpListener, String) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
(listener, url)
}
/// The stand-in, on `listener`, founding groups whose declarations
/// arrive in `index`, the site's, whose identities `site` resolves.
pub fn serve(
listener: tokio::net::TcpListener,
site: &grain::env::Ctx,
index: Arc<grain_serve::index::Index>,
) -> Starter {
let port = listener.local_addr().unwrap().port();
let url = format!("http://127.0.0.1:{port}");
let did = format!("did:web:starter{port}.test");
let dir = tempfile::tempdir().unwrap();
let asked = Arc::new(Mutex::new(Vec::new()));
let sent = Arc::new(Mutex::new(Value::Null));
let founding = Arc::new(Founding {
pds: net::net().pds.trim_end_matches('/').to_string(),
url: url.clone(),
did: did.clone(),
site: site.clone(),
index,
dir: dir.path().to_path_buf(),
asked: asked.clone(),
sent: sent.clone(),
tokens: Mutex::new(HashMap::new()),
seen: Mutex::new(Vec::new()),
});
let router = Router::new()
.route("/.well-known/did.json", get(host_did))
.route("/.well-known/oauth-protected-resource", get(resource))
.route(
"/.well-known/oauth-authorization-server",
get(authorization),
)
.route("/xrpc/com.atproto.server.describeServer", get(describe))
.route("/xrpc/fyi.opensocial.provisionGroup", post(provision))
.route("/xrpc/fyi.opensocial.createSpace", post(create_space))
.fallback(founded)
.with_state(founding);
tokio::spawn(async move { axum::serve(listener, router).await });
Starter {
url,
did,
asked,
sent,
_dir: dir,
}
}
}
crates/grain-serve/tests/web/mod.rs+16−0
View file
crates/grain-serve/tests/moderation.rs+34−0
View file
crates/grain-serve/tests/spaces.rs+96−2
View file
deploy/deploy.sh+3−2
View file
deploy/docker-compose.yml+9−7
View file
deploy/README.md+7−9
View file
lexicons/social/grain/group/createGroup.json+17−1
View file
lexicons/social/grain/group/getGroupHost.json+4−1
View file
lexicons/social/grain/moderation/describeLabels.jsonnew+28−0
View file
lexicons/README.md+6−1
View file