1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
//! Published modules and compositions, by their records: what `atlibs run`/`add` resolve, as a library
//! for hosts that embed atlibs (a CI runner, a server). The CLI adds lockfiles and local builds on top.
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use serde_json::{Map, Value, json};
use crate::compose::{self, Composition};
use crate::crypto::ingress_host;
use crate::host::{DidkeyCall, HostCall, Session};
use crate::oauth::ScopeCall;
use crate::run::Call;
use crate::store::Store;
use crate::transport::{TransportCall, Transports};
use crate::host::{Fetch, blob_sha256, json_response, public_url, xrpc_url};
use crate::BoxFut;
use crate::run::{Module, module_key_ok, module_pin};
use crate::wasm::sha256_hex;
pub const MODULE: &str = "com.fedfork.exec.wasm.module";
pub const CLIENT: &str = "com.fedfork.exec.oauth.clientMetadata";
pub const COMPOSITION: &str = "com.fedfork.exec.composition";
pub const ABI: &str = "emdash-wasm-json-v1";
/// HTTP plus where DIDs and handles resolve (env, so tests and mirrors can point elsewhere).
#[derive(Clone)]
pub struct Net {
pub fetch: Arc<dyn Fetch>,
pub plc: String,
pub resolver: String,
pub allow_private: bool,
}
pub fn split_uri(uri: &str) -> Option<(&str, &str, &str)> {
let mut parts = uri.strip_prefix("at://")?.split('/');
match (parts.next(), parts.next(), parts.next(), parts.next()) {
(Some(did), Some(c), Some(r), None) if did.starts_with("did:") && !c.is_empty() && !r.is_empty() => Some((did, c, r)),
_ => None,
}
}
fn params(pairs: &[(&str, &str)]) -> Map<String, Value> {
pairs.iter().map(|(k, v)| (k.to_string(), json!(v))).collect()
}
impl Net {
/// plc.directory and public.api.bsky.app, public addresses only.
pub fn new(fetch: Arc<dyn Fetch>) -> Self {
Self { fetch, plc: "https://plc.directory".into(), resolver: "https://public.api.bsky.app".into(), allow_private: false }
}
async fn get(&self, nsid: &str, url: &str) -> Result<Value, String> {
let url = public_url(url, self.allow_private)?;
json_response(nsid, &self.fetch.fetch("GET", &url, vec![], None).await?).map_err(|e| e.message)
}
pub async fn did_doc(&self, did: &str) -> Result<Value, String> {
let url = if did.starts_with("did:plc:") {
format!("{}/{did}", self.plc.trim_end_matches('/'))
} else if let Some(host) = did.strip_prefix("did:web:") {
format!("https://{}/.well-known/did.json", host.replace("%3A", ":").replace("%3a", ":"))
} else {
return Err(format!("{did} is not a did:plc or did:web"));
};
self.get(did, &url).await.map_err(|e| format!("resolving {did}: {e}"))
}
pub async fn pds(&self, did: &str) -> Result<String, String> {
let doc = self.did_doc(did).await?;
let pds = doc["service"]
.as_array()
.into_iter()
.flatten()
.find(|s| s["id"] == "#atproto_pds" || s["id"] == format!("{did}#atproto_pds"))
.and_then(|s| s["serviceEndpoint"].as_str())
.ok_or_else(|| format!("{did} has no PDS"))?;
public_url(pds, self.allow_private)
}
/// handle -> DID, checked both ways: the DID document must claim the handle.
// ponytail: resolveHandle at ATLIBS_HANDLE_RESOLVER, no DNS TXT of our own; the alsoKnownAs check
// keeps a lying resolver from substituting another account.
pub async fn resolve_handle(&self, handle: &str) -> Result<String, String> {
let nsid = "com.atproto.identity.resolveHandle";
let found = self.get(nsid, &xrpc_url(&self.resolver, nsid, Some(¶ms(&[("handle", handle)])))?).await.map_err(|e| format!("resolving {handle}: {e}"))?;
let did = found["did"].as_str().ok_or_else(|| format!("{handle} did not resolve"))?.to_string();
let doc = self.did_doc(&did).await?;
let claimed = doc["alsoKnownAs"].as_array().into_iter().flatten().any(|a| a.as_str() == Some(&format!("at://{handle}")));
if !claimed {
return Err(format!("{handle} resolves to {did}, whose DID document does not claim it"));
}
Ok(did)
}
/// DID -> its handle, for display: the DID document's first alsoKnownAs at:// handle, when that handle
/// resolves back to the DID (resolve_handle). None otherwise; integrity never rests on it.
pub async fn handle_of(&self, did: &str) -> Option<String> {
let doc = self.did_doc(did).await.ok()?;
let handle = doc["alsoKnownAs"].as_array()?.iter().find_map(|a| a.as_str()?.strip_prefix("at://").filter(|h| !h.starts_with("did:")).map(String::from))?;
(self.resolve_handle(&handle).await.ok()? == did).then_some(handle)
}
/// (cid, value) of a record; None when the PDS has no such record.
pub async fn record(&self, uri: &str) -> Result<Option<(String, Value)>, String> {
let (did, collection, rkey) = split_uri(uri).ok_or_else(|| format!("{uri} is not an at:// record URI"))?;
let url = xrpc_url(&self.pds(did).await?, "com.atproto.repo.getRecord", Some(¶ms(&[("repo", did), ("collection", collection), ("rkey", rkey)])))?;
let response = self.fetch.fetch("GET", &public_url(&url, self.allow_private)?, vec![], None).await?;
match json_response("com.atproto.repo.getRecord", &response) {
Ok(mut body) => Ok(Some((body["cid"].as_str().unwrap_or_default().to_string(), body["value"].take()))),
Err(e) if response.status == 404 || e.error.as_deref() == Some("RecordNotFound") => Ok(None),
Err(e) => Err(format!("{uri}: {}", e.message)),
}
}
pub async fn blob(&self, did: &str, cid: &str) -> Result<Vec<u8>, String> {
let url = xrpc_url(&self.pds(did).await?, "com.atproto.sync.getBlob", Some(¶ms(&[("did", did), ("cid", cid)])))?;
let response = self.fetch.fetch("GET", &public_url(&url, self.allow_private)?, vec![], None).await?;
if !(200..300).contains(&response.status) {
return Err(format!("blob {cid} of {did} not found ({})", response.status));
}
Ok(response.body)
}
}
/// A session's com.fedfork.exec.oauth.clientMetadata record.
#[derive(Clone)]
pub struct Client {
pub rkey: String,
pub record: Value,
pub uri: Option<String>,
pub cid: Option<String>,
}
/// One module: bytes, its module.json fields, and its session's record.
#[derive(Clone)]
pub struct Lib {
pub key: String,
/// what was asked for (<handle>/<lib>), kept in the lockfile
pub reference: Option<String>,
pub uri: Option<String>,
pub cid: Option<String>,
pub wasm: Vec<u8>,
pub sha256: String,
/// module.json fields; published ones carry their uri and the session's rkey
pub spec: Value,
pub client: Option<Client>,
/// the run identity: its at:// URI, else the wasm's absolute path
pub id: String,
/// Pinned (vendored in a lockfile, or a ref @pin): the directory whose atlibs.lock.json alone may
/// supply its hook defaults (default_lib).
pub pinned: Option<PathBuf>,
}
impl Lib {
/// What its pins match: run::module_pin of its wasm and its record's grants.
pub fn pin(&self) -> String {
module_pin(&self.wasm, &self.spec)
}
}
/// The rkey publish also writes a module's record at, `<rkey>.<first 32 hex of its pin>`, so a republish leaves
/// every pinned version in place. None for a pin that is not 32 to 64 hex (a record CID).
pub fn version_rkey(rkey: &str, pin: &str) -> Option<String> {
((32..=64).contains(&pin.len()) && pin.bytes().all(|b| b.is_ascii_hexdigit())).then(|| format!("{rkey}.{}", pin[..32].to_ascii_lowercase()))
}
/// The rkey publish also writes a composition's record at, `<rkey>.<its record CID>`: what a CID pin (a binding,
/// `@<cid>`) loads, left in place by a republish. None for a CID that is not 1 to 100 letters and digits.
pub fn composition_version(rkey: &str, cid: &str) -> Option<String> {
((1..=100).contains(&cid.len()) && cid.bytes().all(|b| b.is_ascii_alphanumeric())).then(|| format!("{rkey}.{cid}"))
}
/// A composition record: (its CID, it). Pinned by `cid`, the record at composition_version, which must be that CID.
pub async fn composition(net: &Net, uri: &str, cid: Option<&str>) -> Result<(String, Value), String> {
let (did, _, rkey) = split_uri(uri).filter(|(_, c, _)| *c == COMPOSITION).ok_or_else(|| format!("{uri} is not a {COMPOSITION} record"))?;
let at = match cid {
Some(c) => format!("at://{did}/{COMPOSITION}/{}", composition_version(rkey, c).ok_or_else(|| format!("{uri}: {c} is not a record CID"))?),
None => uri.to_string(),
};
let (found, value) = net.record(&at).await?.ok_or_else(|| if cid.is_some() { format!("{uri}: version {} not published ({at} not found)", cid.unwrap_or_default()) } else { format!("{uri} not found") })?;
if let Some(pinned) = cid.filter(|c| *c != found) {
return Err(format!("{uri}: pinned to {pinned}, but it is record {found}"));
}
Ok((found, value))
}
/// A published module: its record, its blob (sha256 checked against the CID), its session record. With a
/// pin (check_pin) the record at version_rkey, whose identity is still `uri`.
pub async fn fetch_lib(net: &Net, uri: &str, reference: Option<&str>, pin: Option<&str>) -> Result<Lib, String> {
let (did, _, rkey) = split_uri(uri).filter(|(_, c, _)| *c == MODULE).ok_or_else(|| format!("{uri} is not a {MODULE} record"))?;
let at = pin.and_then(|p| version_rkey(rkey, p)).map_or_else(|| uri.to_string(), |v| format!("at://{did}/{MODULE}/{v}"));
let (cid, value) = net.record(&at).await?.ok_or_else(|| if at == uri { format!("{uri} not found") } else { format!("{uri}: pin {} not published ({at} not found)", pin.unwrap_or_default()) })?;
if value["abi"] != ABI {
return Err(format!("{uri}: unsupported module ABI {}", value["abi"]));
}
let blob_cid = value.pointer("/module/ref/$link").and_then(Value::as_str);
let expected = blob_sha256(blob_cid)?;
let wasm = net.blob(did, blob_cid.unwrap_or_default()).await?;
let sha256 = sha256_hex(&wasm);
if sha256 != expected {
return Err(format!("{uri}: blob does not match its CID"));
}
let mut spec = spec_of(&value);
spec.insert("uri".into(), json!(uri));
let mut client = None;
if let Some(session_uri) = value.pointer("/session/uri").and_then(Value::as_str) {
let (_, _, session) = split_uri(session_uri).filter(|(_, c, _)| *c == CLIENT).ok_or_else(|| format!("{uri}: session is not a {CLIENT} record"))?;
let (session_cid, record) = net.record(session_uri).await?.ok_or_else(|| format!("{uri}: session record {session_uri} not found"))?;
if Some(session_cid.as_str()) != value.pointer("/session/cid").and_then(Value::as_str) {
eprintln!("warning: {uri}: session record changed since the module was published");
}
spec.insert("session".into(), json!(session));
client = Some(Client { rkey: session.into(), record, uri: Some(session_uri.into()), cid: Some(session_cid) });
}
let lib = Lib { key: rkey.into(), reference: reference.map(String::from), uri: Some(uri.into()), cid: Some(cid), wasm, sha256, spec: Value::Object(spec), client, id: uri.into(), pinned: None };
check_pin(&lib, pin)?;
Ok(lib)
}
/// The module.json fields a host reads, and nothing else: a record (or module.json) cannot choose its
/// own identity or ledger owner (`id`, `owner`, `uri`); those come from where it was loaded.
pub const SPEC_FIELDS: [&str; 12] = ["name", "description", "actions", "config", "calls", "anyRecords", "didkey", "session", "hooks", "implements", "anyHooks", "http"];
pub fn spec_of(value: &Value) -> Map<String, Value> {
SPEC_FIELDS.iter().filter_map(|k| value.get(*k).map(|v| (k.to_string(), v.clone()))).collect()
}
/// The at:// URI a remote ref names (a module, else a composition of that rkey) and its pin.
pub async fn resolve(net: &Net, reference: &str) -> Result<(String, Option<String>), String> {
if reference.starts_with("at://") {
return Ok((reference.to_string(), None));
}
let (path, pin) = match reference.split_once('@') {
Some((p, pin)) => (p, Some(pin.to_string())),
None => (reference, None),
};
let (handle, lib) = path.split_once('/').ok_or_else(|| format!("{reference} is not <handle>/<lib>"))?;
let did = if handle.starts_with("did:") { handle.to_string() } else { net.resolve_handle(handle).await? };
for collection in [MODULE, COMPOSITION] {
let uri = format!("at://{did}/{collection}/{lib}");
if net.record(&uri).await?.is_some() {
return Ok((uri, pin));
}
}
Err(format!("{handle} ({did}) publishes no module or composition named {lib}"))
}
/// How a record is shown to people: `<handle>/<rkey>` (Net::handle_of), else `<did>/<rkey>`; the at:// URI
/// for a collection a `<handle>/<rkey>` does not name (resolve: modules and compositions).
pub async fn named(net: &Net, uri: &str) -> String {
match split_uri(uri) {
Some((did, collection, rkey)) if collection == MODULE || collection == COMPOSITION => format!("{}/{rkey}", net.handle_of(did).await.unwrap_or_else(|| did.into())),
_ => uri.into(),
}
}
/// `@pin`: the record CID, or at least 32 hex characters of its pin (module_pin: the wasm and the record's
/// grants, so a record whose calls or hooks changed no longer matches).
pub fn check_pin(lib: &Lib, pin: Option<&str>) -> Result<(), String> {
let hex = |p: &str| p.bytes().all(|b| b.is_ascii_hexdigit());
match pin {
None => Ok(()),
Some(p) if lib.cid.as_deref() == Some(p) => Ok(()),
Some(p) if hex(p) && p.len() < 32 => Err(format!("{}: pin {p} is too short: give the record CID, or at least 32 hex characters of the pin", lib.key)),
Some(p) if hex(p) && p.len() <= 64 && lib.pin().starts_with(&p.to_ascii_lowercase()) => Ok(()),
Some(p) => Err(format!("{}: pinned to {p}, but it is pin {} (record {})", lib.key, lib.pin(), lib.cid.as_deref().unwrap_or("-"))),
}
}
/// A lib as a runner takes it: its module, with its identity (id, key) and its session's scope.
pub fn module_of(lib: &Lib) -> Result<Module, String> {
let mut module = Module::from_manifest(lib.wasm.clone(), &lib.spec)?;
module.sha256 = Some(lib.sha256.clone());
module.id = Some(lib.id.clone());
module.key = Some(lib.key.clone());
module.scope = lib.client.as_ref().and_then(|c| Some((c.rkey.clone(), c.record["clientMetadata"]["scope"].as_str()?.to_string())));
Ok(module)
}
/// A published composition's instance module: an at:// URI, or a key of its publisher's; at its pin when pinned.
pub async fn instance_lib(net: &Net, reference: &str, publisher: &str, pin: Option<&str>) -> Result<Lib, String> {
if reference.starts_with("at://") {
return fetch_lib(net, reference, None, pin).await;
}
if !module_key_ok(reference) {
return Err(format!("{reference}: a published composition names its publisher's modules by key, or by at:// URI"));
}
fetch_lib(net, &format!("at://{publisher}/{MODULE}/{reference}"), None, pin).await
}
/// Where a composition's module refs come from: (ref, its instance's pin) -> lib.
pub type LibOf = Arc<dyn Fn(String, Option<String>) -> BoxFut<'static, Result<Lib, String>> + Send + Sync>;
/// compose::load with `lib_of` finding each module (pins checked, hook defaults loaded as siblings), and
/// each ref it loaded with its lib.
pub async fn load_with(manifest: &Value, lib_of: LibOf) -> Result<(Composition, Vec<(String, Lib)>), String> {
let found: Arc<Mutex<Vec<(String, Lib)>>> = Arc::default();
let loader: compose::ModuleLoader = {
let found = found.clone();
Arc::new(move |reference: String, pin: Option<String>| {
let (found, lib_of) = (found.clone(), lib_of.clone());
Box::pin(async move {
let lib = lib_of(reference.clone(), pin).await?;
let module = module_of(&lib)?;
found.lock().unwrap().push((reference, lib));
Ok(module)
})
})
};
let comp = compose::load(manifest, &loader).await?;
let found = found.lock().unwrap().clone();
Ok((comp, found))
}
/// The composition record at `uri` (at its version `cid` when given: composition), loaded with its modules from
/// their records. -> (its CID, it, each ref with its lib).
pub async fn published(net: &Net, uri: &str, cid: Option<&str>) -> Result<(String, Composition, Vec<(String, Lib)>), String> {
let (found, value) = composition(net, uri, cid).await?;
let did = split_uri(uri).map(|(d, _, _)| d).unwrap_or_default();
let (net, publisher) = (net.clone(), did.to_string());
let lib_of: LibOf = Arc::new(move |reference: String, pin: Option<String>| {
let (net, publisher) = (net.clone(), publisher.clone());
Box::pin(async move { instance_lib(&net, &reference, &publisher, pin.as_deref()).await })
});
let (comp, libs) = load_with(&value, lib_of).await?;
Ok((found, comp, libs))
}
/// The call one module runs on: XRPC with `session` (held to the module's own session scope when signed
/// in), attest when it lists calls (one that lists none may call anything, so gets no signing key), its
/// did:key (sealed under `name`: its identity, or composition#instance; an OAuth user's under "<did> <name>"),
/// and `transports`.
pub async fn module_call(store: &Store, fetch: Arc<dyn Fetch>, session: Option<Arc<Session>>, transports: &Transports, module: &Module, name: &str) -> Result<Arc<dyn Call>, String> {
let lists_calls = module.calls.as_ref().is_some_and(|c| !c.is_empty());
let attest = (session.is_some() && lists_calls).then(|| store.clone());
let mut call: Arc<dyn Call> = Arc::new(HostCall { fetch: fetch.clone(), session: session.clone(), attest });
if session.is_some() {
let key = module.key.clone().unwrap_or_else(|| name.to_string());
call = Arc::new(ScopeCall { module: key, session: module.scope.clone(), inner: call });
}
if let (Some(didkey), true) = (&module.didkey, lists_calls) {
let name = match &session {
Some(s) if s.is_oauth() => format!("{} {name}", s.did()),
_ => name.to_string(),
};
let key = store.didkey(&name).await?;
let services: Vec<String> = didkey.get("services").and_then(Value::as_array).map(|s| s.iter().filter_map(|x| x.as_str().map(String::from)).collect()).unwrap_or_default();
for service in &services {
eprintln!("{}: https://{}/", key.did, ingress_host(&key.did, service));
}
call = Arc::new(DidkeyCall { key, services, fetch, inner: call, store: Some(store.clone()) });
}
Ok(Arc::new(TransportCall { transports: transports.clone(), listed: lists_calls, inner: call }))
}
