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
// atlibs run: one run of a module entry (or a composition instance's), as javascript/src/server/cli.mjs.
// atlibs run <handle>/<lib> | <at://did/com.fedfork.exec.wasm.module/rkey> <entry> ['<json>' | - | @file]
// atlibs run <name in atlibs.lock.json> <entry> ['<json>']
// atlibs run <module dir | name under examples/rust> <entry> ['<json>'] (module.json + the cargo wasm build)
// atlibs run <module.wasm> <module.json> <entry> ['<json>']
// atlibs run <stage.json | atlibs.lock.json> <module> <entry> ['<json>']
// atlibs run <composition .json | composition in the lockfile | <handle>/<composition>> <instance> <entry> ['<json>']
// [--handler <name>=<spec> ...] [--allow-exec] the composition's declared handlers (handlers.rs)
// [--params '<json>' | @file] {inputs, secrets, host}: its parameters (atlibs_host::params::bind)
// '<json>' may be `-` (read stdin) or `@file`: a `stdin` secret stays out of argv. Prints runModule's
// { entry, result, error?, events, calls } as JSON on stdout; a run error is also one line on stderr
// and exit status 1. Ctrl-C ends a run and still prints it.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::collections::HashSet;
use std::sync::Arc;
use atlibs_host::compose::{self, Composition};
use atlibs_host::host::{Fetch, Session};
use atlibs_host::run::{self, Call, Ctx, Loader, Module, Options, Runner, run_module};
use atlibs_host::store::Store;
use atlibs_host::transport::Transports;
use atlibs_host::wasm::{Bound, WALL_TIME_MS, Wasm};
use atlibs_host::{Signal, pretty};
use serde_json::{Map, Value, json};
use crate::registry::{self, Lib, Net, Ref};
use crate::{Cli, env};
/// The NSIDs this build's transports serve (cargo features), for loading before a Host exists (refs,
/// compositions): the transports keyed under the state dir this process would use (Host::provides is the
/// same set, from the Host's own transports).
pub fn provides() -> HashSet<String> {
atlibs_host::transport::provides(&crate::transports(&crate::state_dir()))
}
/// What a ref runs: one module (pinned: the dir whose lockfile alone supplies its hook defaults, Lib::pinned),
/// or a composition (its identity is its file, or its at:// URI).
pub enum Target {
Module(Module, Option<PathBuf>),
Composition { id: String, comp: Composition },
}
impl Target {
pub fn is_composition(&self) -> bool {
matches!(self, Target::Composition { .. })
}
}
pub use atlibs_host::registry::module_of;
/// The ref at the front of `words` (registry::resolve_ref; lockfile names in ./atlibs.lock.json) as what
/// runs, and the words after it.
pub async fn target<'a>(net: &Net, words: &'a [String]) -> Result<(Target, &'a [String]), String> {
let (found, rest) = registry::resolve_ref(net, Path::new("."), words).await?;
Ok((target_of(found)?, rest))
}
pub fn target_of(found: Ref) -> Result<Target, String> {
match found {
Ref::Module(lib) => Ok(Target::Module(module_of(&lib)?, lib.pinned)),
// composition_libs loaded it (hook defaults added); what is left is the transports this build serves
Ref::Composition { id, comp, .. } => compose::check(&comp, &provides()).map(|_| Target::Composition { id, comp }),
Ref::Client(c) => Err(format!("{} is a clientMetadata record: no module to run", c.uri.unwrap_or(c.rkey))),
}
}
/// The entries a module.json's actions name (`atlibs run <ref>` with no entry, serve's index).
pub fn actions(lib: &Lib) -> Vec<&str> {
lib.spec["actions"].as_array().into_iter().flatten().filter_map(|a| a["entry"].as_str()).collect()
}
/// What `atlibs run <ref>` with no entry prints: the entries its module.json actions name.
fn entries(found: &Ref) -> String {
let actions = |lib: &Lib| actions(lib).join(", ");
match found {
Ref::Module(lib) => format!("{} entries: {}", lib.key, actions(lib)),
Ref::Composition { manifest, libs, .. } => manifest["instances"]
.as_array()
.into_iter()
.flatten()
.map(|inst| format!("{} ({}): {}", inst["name"].as_str().unwrap_or_default(), inst["module"].as_str().unwrap_or_default(), libs.iter().find(|l| inst["module"] == l.key.as_str()).map(actions).unwrap_or_default()))
.collect::<Vec<_>>()
.join("\n"),
Ref::Client(_) => String::new(),
}
}
pub fn input_of(json: Option<&String>) -> Result<Map<String, Value>, String> {
let json = json.map_or("{}", String::as_str);
match serde_json::from_str::<Value>(json) {
Ok(Value::Object(input)) => Ok(input),
Ok(_) => Err("input must be a JSON object".into()),
// never the text itself: it may carry a secret (stdin)
Err(e) => Err(format!("input is not JSON (line {}, column {})", e.line(), e.column())),
}
}
/// The CLI input word: '<json>', `-` (stdin) or `@file`, so a secret (a `stdin` field) stays out of argv.
pub fn input_arg(word: Option<&String>) -> Result<Option<String>, String> {
let read = |what: &str, got: std::io::Result<String>| got.map_err(|e| format!("input {what}: {e}"));
match word.map(String::as_str) {
Some("-") => read("-", std::io::read_to_string(std::io::stdin())).map(Some),
Some(w) if w.starts_with('@') => read(w, std::fs::read_to_string(&w[1..])).map(Some),
_ => Ok(word.cloned()),
}
}
/// The signed-in host every run shares: store, HTTP, session, wasm engine.
#[derive(Clone)]
pub struct Host {
pub store: Store,
pub fetch: Arc<dyn Fetch>,
pub session: Option<Arc<Session>>,
pub wasm: Arc<Wasm>,
pub budget: Option<Value>,
/// `--handler` flags: the caller handlers a composition run gets (handlers.rs)
pub handlers: Vec<(String, crate::handlers::Spec)>,
/// this build's transports (crate::transports), made once, and the NSIDs they serve (required hooks)
pub transports: Transports,
pub provides: Arc<HashSet<String>>,
/// ATLIBS_HTTP: the origins com.fedfork.exec.http.fetch may reach, with the headers this host adds
pub http: Option<Arc<run::Http>>,
/// ATLIBS_HELD: { name: value }, the held values each run starts with (a composition's root's)
pub held: Option<Map<String, Value>>,
}
/// A JSON env variable, or @file of it.
fn env_json(name: &str) -> Result<Option<Value>, String> {
let Some(text) = env(name) else { return Ok(None) };
let text = match text.strip_prefix('@') {
Some(path) => std::fs::read_to_string(path).map_err(|e| format!("{name} {path}: {e}"))?,
None => text,
};
serde_json::from_str(&text).map(Some).map_err(|e| format!("{name}: {e}"))
}
impl Host {
pub async fn new(cli: &Cli) -> Result<Self, String> {
let budget = match env("ATLIBS_BUDGET") {
Some(b) => Some(serde_json::from_str::<Value>(&b).map_err(|e| format!("ATLIBS_BUDGET: {e}"))?),
None => None,
};
let transports = crate::transports(&cli.state_dir);
let provides = Arc::new(atlibs_host::transport::provides(&transports));
// JSON, or @file of it; its own client, following no redirects (an injected key must not travel)
let http = match env_json("ATLIBS_HTTP")? {
Some(value) => {
let client = reqwest::Client::builder().user_agent("atlibs").redirect(reqwest::redirect::Policy::none()).build().map_err(|e| e.to_string())?;
let mut http = run::Http::from_json(Arc::new(client), &value).map_err(|e| format!("ATLIBS_HTTP: {e}"))?;
// "iroh://*" in ATLIBS_HTTP: that transport's URLs
http.schemes = atlibs_host::transport::fetchers(&transports);
Some(Arc::new(http))
}
None => None,
};
let held = match env_json("ATLIBS_HELD")? {
None => None,
Some(Value::Object(held)) => Some(held),
Some(_) => return Err("ATLIBS_HELD must be { name: value }".into()),
};
Ok(Self { store: cli.store()?.clone(), fetch: cli.fetch.clone(), session: cli.session().await?, wasm: Wasm::new(WALL_TIME_MS), budget, handlers: vec![], transports, provides, http, held })
}
/// "no entry <entry> (<its entries>)" when the module (a composition's: `instance`'s) does not export
/// `entry`; None also when that can't be told here (no such instance, not wasm: the run says why).
pub fn no_entry(&self, target: &Target, instance: Option<&str>, entry: &str) -> Option<String> {
let module = match (target, instance) {
(Target::Module(module, _), _) => module,
(Target::Composition { comp, .. }, Some(instance)) => comp.modules.get(instance)?,
_ => return None,
};
let entries = self.wasm.entries(&module.bytes).ok()?;
(!entries.iter().any(|e| e == entry)).then(|| format!("no entry {entry} ({})", entries.join(", ")))
}
/// The `call` for one module: hostCall (held to its session's scope when signed in), with attest when the module lists calls (a module that
/// lists none may call anything, so it gets no signing key), its did:key (sealed under `name`:
/// the module's identity, or composition#instance), and the transports this CLI was built with.
pub async fn call_for(&self, module: &Module, name: &str) -> Result<Arc<dyn Call>, String> {
atlibs_host::registry::module_call(&self.store, self.fetch.clone(), self.session.clone(), &self.transports, module, name).await
}
/// A hook default of `module`, as registry::default_lib resolves it (`pinned`: only from
/// that dir's lockfile), with its own call (call_for: its calls, its attest and did:key, never `module`'s).
pub fn hook_loader(&self, module: &Module, pinned: Option<PathBuf>) -> Loader {
let (host, id) = (self.clone(), module.id.clone().unwrap_or_default());
Arc::new(move |key: String| {
let (host, id, pinned) = (host.clone(), id.clone(), pinned.clone());
Box::pin(async move {
let lib = registry::default_lib(®istry::net_from_env(host.fetch.clone()), &id, pinned.as_deref(), &key).await?;
let module = module_of(&lib)?;
let call = host.call_for(&module, &lib.id).await?;
let runner: Arc<dyn Runner> = Arc::new(Bound { wasm: host.wasm.clone(), bytes: module.bytes.clone(), sha256: module.sha256.clone() });
Ok((module, runner, call))
})
})
}
/// One run: a module's entry, or `instance`'s entry in a composition. The signed-in DID is input.did.
pub async fn run(&self, target: &Target, instance: Option<&str>, entry: &str, input: Map<String, Value>, signal: Signal) -> Result<Value, String> {
let mut with_did = Map::new();
if let Some(s) = &self.session {
with_did.insert("did".into(), json!(s.did()));
}
with_did.extend(input);
let options = Options { signal: Some(signal), budget: self.budget.clone(), provides: Some(self.provides.clone()), http: self.http.clone(), held: self.held.clone(), ..Default::default() };
let ledger = Some(self.store.ledger("managed_records"));
match target {
Target::Module(..) if !self.handlers.is_empty() => Err("--handler supplies a composition's handlers; this ref is one module".into()),
Target::Module(module, pinned) => {
// did:keys are keyed by the module's identity (its at:// URI or absolute path), not the name it
// gives itself: a module must not claim another's key (and ingress host) by naming itself so.
let call = self.call_for(module, module.id.as_deref().unwrap_or_default()).await?;
let ctx = Ctx { runner: Arc::new(Bound { wasm: self.wasm.clone(), bytes: module.bytes.clone(), sha256: module.sha256.clone() }), ledger, now: None };
let options = Options { load: Some(self.hook_loader(module, pinned.clone())), ..options };
Ok(run_module(&ctx, module, entry, &with_did, call, options).await)
}
Target::Composition { id, comp } => {
let instance = instance.ok_or("a composition runs one of its instances: <ref> <instance> <entry>")?;
let handlers = crate::handlers::build(self, &comp.manifest, &self.handlers)?;
// each instance's did:key is sealed under <composition id>#<instance>, so an instance name
// cannot claim another module's key
let mut calls = HashMap::new();
for (name, module) in &comp.modules {
calls.insert(name.clone(), self.call_for(module, &format!("{id}#{name}")).await?);
}
let comp = Arc::new(Composition { manifest: comp.manifest.clone(), modules: comp.modules.clone() });
let call_for: compose::CallFor = Arc::new(move |name: &str| calls.get(name).cloned());
compose::run_compose(comp, self.wasm.clone(), ledger, instance, entry, with_did, call_for, handlers, None, options).await
}
}
}
}
pub async fn main(cli: &Cli, argv: &[String], signal: Signal) -> Result<(), String> {
let usage = || crate::usage("run");
let args = crate::Args::parse(argv, &["allow-exec"])?;
args.only(&["handler", "allow-exec", "params"], "run")?;
let argv = &args.pos[..];
if argv.is_empty() {
return Err(usage());
}
let (found, rest) = registry::resolve_ref(&cli.net, Path::new("."), argv).await?;
let listing = entries(&found);
let mut target = target_of(found)?;
let (instance, rest) = if target.is_composition() { (rest.first().map(String::as_str), rest.get(1..).unwrap_or_default()) } else { (None, rest) };
let (Some(entry), true) = (rest.first(), (1..=2).contains(&rest.len())) else {
return Err(if rest.is_empty() { format!("{}\n{listing}", usage()) } else { usage() });
};
// a composition's parameters, bound for this run (listing its entries needs none)
if let Target::Composition { comp, .. } = &mut target {
let given = match input_arg(args.flag("params").map(String::from).as_ref())? {
Some(text) => serde_json::from_str(&text).map_err(|_| "--params is not JSON".to_string())?,
None => json!({}),
};
comp.manifest = atlibs_host::params::bind(&comp.manifest, &atlibs_host::params::Bind::from_json(&given)?)?;
} else if args.has("params") {
return Err("--params binds a composition's parameters; this ref is one module".into());
}
let input = input_of(input_arg(rest.get(1))?.as_ref())?;
let mut host = Host::new(cli).await?;
host.handlers = crate::handlers::parse(cli, &args.all("handler"), args.has("allow-exec")).await?;
if let Some(no) = host.no_entry(&target, instance, entry) {
return Err(no);
}
let ran = host.run(&target, instance, entry, input, signal).await?;
out!("{}", pretty(&ran));
// the result is printed either way; a run error is the exit status (1) and one stderr line
match ran.get("error").filter(|e| !e.is_null()) {
Some(e) => Err(format!("{entry}: {}", e.as_str().map_or_else(|| e.to_string(), String::from))),
None => Ok(()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::fake::Pds;
use crate::registry::tests::tempdir;
#[tokio::test]
async fn refs_resolve_to_targets_with_their_identity() {
let pds = Pds::new();
let uri = pds.publish_module("alice.test", "echo", b"\0asm", None);
let cli = pds.cli(&tempdir("run"));
let args: Vec<String> = ["alice.test/echo", "watch", "{}"].iter().map(|s| s.to_string()).collect();
let (found, rest) = target(&cli.net, &args).await.unwrap();
let Target::Module(module, _) = found else { panic!("a module") };
assert_eq!(module.id.as_deref(), Some(uri.as_str()), "did:keys and the ledger key on the at:// URI");
assert_eq!(rest, &args[1..]);
let wasm: Vec<String> = vec!["/nope.wasm".into()];
assert!(target(&cli.net, &wasm).await.err().unwrap().contains("needs a second word"));
let missing: Vec<String> = vec!["no-such-module".into()];
assert!(target(&cli.net, &missing).await.err().unwrap().contains("no module no-such-module"));
assert!(input_of(Some(&"[1]".to_string())).unwrap_err().contains("JSON object"));
let bad = input_of(Some(&r#"{"stdin":"secret-token""#.to_string())).unwrap_err();
assert!(bad.starts_with("input is not JSON") && !bad.contains("secret"), "{bad}");
let file = tempdir("run-input").join("in.json");
std::fs::write(&file, r#"{"stdin":"tok"}"#).unwrap();
let read = input_arg(Some(&format!("@{}", file.display()))).unwrap();
assert_eq!(input_of(read.as_ref()).unwrap()["stdin"], "tok");
assert_eq!(input_arg(Some(&"{}".to_string())).unwrap().as_deref(), Some("{}"));
}
}
