uqbar-dao/nectar
1
1
Fork 0
mirror of https://github.com/uqbar-dao/nectar.git synced 2024-11-23 03:44:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into jf/invite

Browse Source
This commit is contained in:
realisation 2023-10-18 17:27:19 -04:00
commit 75a9da069f
48 changed files with 1348 additions and 661 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -1,4 +1,4 @@
Last updated: 10/02/23
Last updated: 10/16/23
## Setup
### Building components
@ -9,8 +9,11 @@ Last updated: 10/02/23
git clone git@github.com:uqbar-dao/uqbar.git
git clone git@github.com:uqbar-dao/redb.git
Make sure the two repos are next to each other in your directory structure.
# Get some stuff so we can build wasm.
cd uqbar
cargo install wasm-tools
rustup install nightly
rustup target add wasm32-wasi
@ -23,7 +26,6 @@ git submodule update --init --recursive
# Build the runtime, along with a number of booted-at-startup WASM modules including terminal and key_value
# OPTIONAL: --release flag
cd uqbar
cargo +nightly build --release
# Create the home directory for your node

13
build.rs
View File

@ -36,6 +36,8 @@ where
fn build_app(target_path: &str, name: &str, parent_pkg_path: Option<&str>) {
let pwd = std::env::current_dir().unwrap();
println!("cargo:warning=building {}", target_path);
// Copy in newly-made wit IF old one is outdated
if file_outdated(
format!("{}/wit/", pwd.display()),
@ -124,7 +126,7 @@ fn main() {
}
// only execute if one of the modules has source code changes
const WASI_APPS: [&str; 9] = [
"app_tracker",
"app_store",
"chess",
"homepage",
"http_bindings",
@ -181,10 +183,11 @@ fn main() {
for entry in std::fs::read_dir(&modules_dir).unwrap() {
let entry_path = entry.unwrap().path();
let package_name = entry_path.file_name().unwrap().to_str().unwrap();
// // NOT YET building KV, waiting for deps to be ready
// if package_name == "key_value" {
// continue;
// }
// NOT YET building KV, waiting for deps to be ready
if package_name == "key_value" {
continue;
}
// If Cargo.toml is present, build the app
let parent_pkg_path = format!("{}/pkg", entry_path.display());

0
modules/app_tracker/Cargo-component.lock → modules/app_store/Cargo-component.lock
View File

68
modules/app_tracker/Cargo.lock → modules/app_store/Cargo.lock generated
View File

@ -9,12 +9,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4668cab20f66d8d020e1fbc0ebe47217433c1b6c8f2040faf858554e394ace6"
[[package]]
name = "app_tracker"
name = "app_store"
version = "0.1.0"
dependencies = [
"anyhow",
"bincode",
"cargo-component-bindings",
"rand",
"serde",
"serde_json",
"wit-bindgen",
@ -65,6 +66,12 @@ dependencies = [
"wit-component",
]
[[package]]
name = "cfg-if"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "equivalent"
version = "1.0.1"
@ -80,6 +87,17 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "getrandom"
version = "0.2.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be4136b2a15dd319360be1c07d9933517ccf0be8f16bf62a3bee4f0d618df427"
dependencies = [
"cfg-if",
"libc",
"wasi",
]
[[package]]
name = "hashbrown"
version = "0.14.0"
@ -134,6 +152,12 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "884e2677b40cc8c339eaefcb701c32ef1fd2493d71118dc0ca4b6a736c93bd67"
[[package]]
name = "libc"
version = "0.2.149"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a08173bc88b7955d1b3145aa561539096c421ac8debde8cbc3612ec635fee29b"
[[package]]
name = "log"
version = "0.4.20"
@ -152,6 +176,12 @@ version = "2.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b2a4787296e9989611394c33f193f676704af1686e70b8f8033ab5ba9a35a94"
[[package]]
name = "ppv-lite86"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
[[package]]
name = "proc-macro2"
version = "1.0.66"
@ -181,6 +211,36 @@ dependencies = [
"proc-macro2",
]
[[package]]
name = "rand"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
dependencies = [
"libc",
"rand_chacha",
"rand_core",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core",
]
[[package]]
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom",
]
[[package]]
name = "ryu"
version = "1.0.15"
@ -324,6 +384,12 @@ version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
[[package]]
name = "wasi"
version = "0.11.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
[[package]]
name = "wasm-encoder"
version = "0.32.0"

3
modules/app_tracker/Cargo.toml → modules/app_store/Cargo.toml
View File

@ -1,5 +1,5 @@
[package]
name = "app_tracker"
name = "app_store"
version = "0.1.0"
edition = "2021"
@ -14,6 +14,7 @@ lto = true
anyhow = "1.0"
bincode = "1.3.3"
cargo-component-bindings = { git = "https://github.com/bytecodealliance/cargo-component" }
rand = "0.8.5"
serde = {version = "1.0", features = ["derive"] }
serde_json = "1.0"
wit-bindgen = { version = "0.11.0", default_features = false }

8
modules/app_tracker/pkg/manifest.json → modules/app_store/pkg/manifest.json
View File

@ -1,7 +1,7 @@
[
{
"process_name": "app_tracker",
"process_wasm_path": "/app_tracker.wasm",
"process_name": "main",
"process_wasm_path": "/app_store.wasm",
"on_panic": "Restart",
"request_networking": true,
"request_messaging": [
@ -16,6 +16,6 @@
"kernel:sys:uqbar",
"eth_rpc:sys:uqbar"
],
"grant_messaging": []
"public": false
}
]
]

5
modules/app_store/pkg/metadata.json Normal file
View File

@ -0,0 +1,5 @@
{
"package": "app_store",
"publisher": "uqbar",
"description": "A package manager + app store. This JSON field is optional and you can add whatever you want in addition to this."
}

0
modules/app_tracker/src/kernel_types.rs → modules/app_store/src/kernel_types.rs
View File

512
modules/app_store/src/lib.rs Normal file
View File

@ -0,0 +1,512 @@
cargo_component_bindings::generate!();
use bindings::{
component::uq_process::types::*, get_capability, get_payload, print_to_terminal, receive,
send_request, send_response, Guest,
};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
#[allow(dead_code)]
mod kernel_types;
use kernel_types as kt;
#[allow(dead_code)]
mod process_lib;
use process_lib::PackageId;
mod transfer_lib;
struct Component;
// #[derive(Serialize, Deserialize)]
// struct AppState {
// // TODO this should mirror onchain listing
// pub name: String,
// pub owner: NodeId,
// pub desc: String,
// pub website: Option<String>,
// pub versions: Vec<(u32, String)>, // TODO
// }
#[derive(Serialize, Deserialize)]
struct AppTrackerState {
pub mirrored_packages: HashSet<PackageId>,
pub requested_packages: HashMap<PackageId, NodeId>, // who we're expecting it from
}
#[derive(Debug, Serialize, Deserialize)]
pub enum AppTrackerRequest {
New {
package: PackageId,
mirror: bool,
},
NewFromRemote {
package_id: PackageId,
install_from: NodeId,
},
Install {
package: PackageId,
},
}
#[derive(Debug, Serialize, Deserialize)]
pub enum AppTrackerResponse {
New { package: String },
NewFromRemote { package_id: PackageId },
Install { package: String },
Error { error: String },
}
#[derive(Debug, Serialize, Deserialize)]
pub struct PackageMetadata {
pub package: String,
pub publisher: String,
pub desc: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct PackageManifestEntry {
pub process_name: String,
pub process_wasm_path: String,
pub on_panic: kt::OnPanic,
pub request_networking: bool,
pub request_messaging: Vec<String>,
pub public: bool,
}
fn parse_command(
our: &Address,
source: &Address,
request_string: String,
state: &mut AppTrackerState,
) -> anyhow::Result<Option<AppTrackerResponse>> {
match serde_json::from_str(&request_string)? {
// create a new package based on local payload
AppTrackerRequest::New { package, mirror } => {
if our.node != source.node {
return Err(anyhow::anyhow!("new package request from non-local node"));
}
let Some(mut payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
let vfs_address = Address {
node: our.node.clone(),
process: ProcessId::from_str("vfs:sys:uqbar")?,
};
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::New,
})?),
None,
None,
5,
)?;
// add zip bytes
payload.mime = Some("application/zip".to_string());
let _ = process_lib::send_and_await_response(
&vfs_address,
true,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::Add {
full_path: package.to_string(),
entry_type: kt::AddEntryType::ZipArchive,
},
})?),
None,
Some(&payload),
5,
)?;
// save the zip file itself in VFS for sharing with other nodes
// call it <package>.zip
let _ = process_lib::send_and_await_response(
&vfs_address,
true,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::Add {
full_path: format!("/{}.zip", package.to_string()),
entry_type: kt::AddEntryType::NewFile,
},
})?),
None,
Some(&payload),
5,
)?;
// if mirror, save in our state
if mirror {
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::GetEntry("/metadata.json".into()),
})?),
None,
None,
5,
)?;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no metadata payload"));
};
let metadata = String::from_utf8(payload.bytes)?;
let metadata = serde_json::from_str::<PackageMetadata>(&metadata)?;
state
.mirrored_packages
.insert(PackageId::new(&metadata.package, &metadata.publisher));
process_lib::set_state::<AppTrackerState>(&state);
}
Ok(Some(AppTrackerResponse::New { package: package.to_string() }))
}
// if we are the source, forward to install_from target.
// if we install_from, respond with package if we have it
AppTrackerRequest::NewFromRemote {
package_id,
install_from,
} => {
if our.node == source.node {
let _ = send_request(
&Address {
node: install_from.clone(),
process: our.process.clone(),
},
&Request {
inherit: true,
expects_response: Some(5), // TODO
ipc: Some(serde_json::to_string(&AppTrackerRequest::NewFromRemote {
package_id: package_id.clone(),
install_from: install_from.clone(),
})?),
metadata: None,
},
None,
None,
);
state.requested_packages.insert(package_id, install_from);
process_lib::set_state::<AppTrackerState>(&state);
Ok(None)
} else if our.node == install_from {
let Some(_mirror) = state.mirrored_packages.get(&package_id) else {
return Ok(Some(AppTrackerResponse::Error { error: "package not mirrored here!".into() }))
};
// get the .zip from VFS and attach as payload to response
let vfs_address = Address {
node: our.node.clone(),
process: ProcessId::from_str("vfs:sys:uqbar")?,
};
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package_id.to_string(),
action: kt::VfsAction::GetEntry(format!("/{}.zip", package_id.to_string())),
})?),
None,
None,
5,
)?;
Ok(Some(AppTrackerResponse::NewFromRemote { package_id }))
} else {
// TODO what to do here?
Ok(None)
}
}
AppTrackerRequest::Install { package } => {
if our.node != source.node {
return Err(anyhow::anyhow!("install request from non-local node"));
}
let vfs_address = Address {
node: our.node.clone(),
process: ProcessId::from_str("vfs:sys:uqbar")?,
};
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::GetEntry("/manifest.json".into()),
})?),
None,
None,
5,
)?;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
let manifest = String::from_utf8(payload.bytes)?;
let manifest = serde_json::from_str::<Vec<PackageManifestEntry>>(&manifest)?;
for entry in manifest {
let path = if entry.process_wasm_path.starts_with("/") {
entry.process_wasm_path
} else {
format!("/{}", entry.process_wasm_path)
};
let (_, hash_response) = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::GetHash(path.clone()),
})?),
None,
None,
5,
)?;
let Message::Response((Response { ipc: Some(ipc), .. }, _)) = hash_response else {
return Err(anyhow::anyhow!("bad vfs response"));
};
let kt::VfsResponse::GetHash(Some(hash)) = serde_json::from_str(&ipc)? else {
return Err(anyhow::anyhow!("no hash in vfs"));
};
// build initial caps
let mut initial_capabilities: HashSet<kt::SignedCapability> = HashSet::new();
if entry.request_networking {
let Some(networking_cap) = get_capability(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar")?,
},
&"\"network\"".to_string(),
) else {
return Err(anyhow::anyhow!("app-store: no net cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(networking_cap));
}
let Some(read_cap) = get_capability(
&vfs_address.clone(),
&serde_json::to_string(&serde_json::json!({
"kind": "read",
"drive": package.to_string(),
}))?,
) else {
return Err(anyhow::anyhow!("app-store: no read cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(read_cap));
let Some(write_cap) = get_capability(
&vfs_address.clone(),
&serde_json::to_string(&serde_json::json!({
"kind": "write",
"drive": package.to_string(),
}))?,
) else {
return Err(anyhow::anyhow!("app-store: no write cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(write_cap));
for process_name in &entry.request_messaging {
let Ok(parsed_process_id) = ProcessId::from_str(&process_name) else {
// TODO handle arbitrary caps here
continue;
};
let Some(messaging_cap) = get_capability(
&Address {
node: our.node.clone(),
process: parsed_process_id.clone(),
},
&"\"messaging\"".into()
) else {
print_to_terminal(0, &format!("app-store: no cap for {} to give away!", process_name));
continue;
};
initial_capabilities.insert(kt::de_wit_signed_capability(messaging_cap));
}
let process_id = format!("{}:{}", entry.process_name, package.to_string());
let Ok(parsed_new_process_id) = ProcessId::from_str(&process_id) else {
return Err(anyhow::anyhow!("app-store: invalid process id!"));
};
let _ = process_lib::send_request(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar")?,
},
false,
Some(serde_json::to_string(&kt::KernelCommand::KillProcess(
kt::ProcessId::de_wit(parsed_new_process_id.clone()),
))?),
None,
None,
None,
);
// kernel start process takes bytes as payload + wasm_bytes_handle...
// reconsider perhaps
let (_, _bytes_response) = process_lib::send_and_await_response(
&vfs_address,
false,
Some(serde_json::to_string(&kt::VfsRequest {
drive: package.to_string(),
action: kt::VfsAction::GetEntry(path),
})?),
None,
None,
5,
)?;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no wasm bytes payload."));
};
let _ = process_lib::send_and_await_response(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar")?,
},
false,
Some(serde_json::to_string(&kt::KernelCommand::StartProcess {
id: kt::ProcessId::de_wit(parsed_new_process_id),
wasm_bytes_handle: hash,
on_panic: entry.on_panic,
initial_capabilities,
public: entry.public,
})?),
None,
Some(&payload),
5,
)?;
}
Ok(Some(AppTrackerResponse::Install { package: package.to_string() }))
}
}
}
impl Guest for Component {
fn init(our: Address) {
assert_eq!(our.process.to_string(), "main:app_store:uqbar");
// grant messaging caps to http_bindings and terminal
let Some(our_messaging_cap) = bindings::get_capability(
&our,
&"\"messaging\"".into()
) else {
panic!("missing self-messaging cap!")
};
bindings::share_capability(
&ProcessId::from_str("http_bindings:http_bindings:uqbar").unwrap(),
&our_messaging_cap,
);
bindings::share_capability(
&ProcessId::from_str("terminal:terminal:uqbar").unwrap(),
&our_messaging_cap,
);
print_to_terminal(0, &format!("app_store main proc: start"));
let mut state = process_lib::get_state::<AppTrackerState>().unwrap_or(AppTrackerState {
mirrored_packages: HashSet::new(),
requested_packages: HashMap::new(),
});
loop {
let (source, message) = match receive() {
Ok((source, message)) => (source, message),
Err((error, _context)) => {
print_to_terminal(0, &format!("net error: {:?}", error.kind));
continue;
}
};
match message {
Message::Request(Request {
ipc,
expects_response,
metadata,
..
}) => {
let Some(command) = ipc else {
continue;
};
match parse_command(&our, &source, command, &mut state) {
Ok(response) => {
if let Some(_) = expects_response {
let _ = send_response(
&Response {
inherit: true,
ipc: Some(serde_json::to_string(&response).unwrap()),
metadata,
},
None, // payload will be attached here if created in parse_command
);
};
}
Err(e) => {
print_to_terminal(0, &format!("app-store: got error {}", e));
if let Some(_) = expects_response {
let error = AppTrackerResponse::Error {
error: format!("{}", e),
};
let _ = send_response(
&Response {
inherit: false,
ipc: Some(serde_json::to_string(&error).unwrap()),
metadata,
},
None,
);
};
}
}
}
Message::Response((response, _)) => {
print_to_terminal(0, &format!("app-store: got response {:?}", response));
// only expecting NewFromRemote for apps we've requested
match serde_json::from_str(&response.ipc.unwrap_or_default()) {
Ok(AppTrackerResponse::NewFromRemote { package_id }) => {
if let Some(install_from) = state.requested_packages.remove(&package_id)
{
if install_from == source.node {
// auto-take zip from payload and request ourself with New
let _ = send_request(
&our,
&Request {
inherit: true, // will inherit payload!
expects_response: None,
ipc: Some(
serde_json::to_string(&AppTrackerRequest::New {
package: package_id,
mirror: true,
})
.unwrap(),
),
metadata: None,
},
None,
None,
);
} else {
print_to_terminal(
0,
&format!(
"app-store: got install response from bad source: {}",
install_from
),
);
}
}
}
err => {
print_to_terminal(
0,
&format!("app-store: got unexpected response {:?}", err),
);
}
}
}
}
}
}
}

0
modules/app_tracker/src/process_lib.rs → modules/app_store/src/process_lib.rs
View File

227
modules/app_store/src/transfer_lib.rs Normal file
View File

@ -0,0 +1,227 @@
use super::bindings::component::uq_process::types::*;
use crate::bindings::{get_payload, receive, send_request, send_response};
use serde::{Deserialize, Serialize};
#[derive(Debug)]
pub enum TransferError {
// in all errors, u64 is number of bytes successfully transferred
TargetOffline(u64),
TargetTimeout(u64),
TargetRejected(u64),
SourceFailed(u64),
}
#[derive(Debug, Serialize, Deserialize)]
pub enum TransferMetadata {
Begin {
file_name: String,
file_size: u64,
total_chunks: u64,
},
}
pub fn transfer(
to_addr: Address,
bytes: Vec<u8>,
max_timeout: u64,
) -> (
Result<(), TransferError>,
Vec<Result<(Address, Message), (SendError, Option<Context>)>>,
) {
let transfer_context_id: u64 = rand::random();
let mut bytes_remaining: u64 = bytes.len() as u64;
let mut offset: u64 = 0;
let mut chunk_size: u64 = 1048576; // 1MB
let mut chunks_sent = 0;
let total_chunks = (bytes.len() as f64 / chunk_size as f64).ceil() as u64;
loop {
chunks_sent += 1;
if bytes_remaining < chunk_size {
chunk_size = bytes_remaining;
}
let payload = Payload {
mime: None,
bytes: bytes[offset as usize..offset as usize + chunk_size as usize].to_vec(),
};
send_request(
&to_addr,
&Request {
inherit: false,
expects_response: Some(max_timeout),
ipc: None,
metadata: Some(if chunks_sent == 1 {
serde_json::to_string(&TransferMetadata::Begin {
file_name: "test".to_string(),
file_size: bytes.len() as u64,
total_chunks,
})
.unwrap()
} else {
chunks_sent.to_string()
}),
},
Some(&&transfer_context_id.to_string()),
Some(&payload),
);
bytes_remaining -= chunk_size;
offset += chunk_size;
if bytes_remaining == 0 {
break;
}
}
let mut chunks_confirmed = 0;
let mut non_transfer_message_queue = Vec::new();
loop {
let next = receive();
if let Err((send_error, context)) = &next {
match context {
Some(_) => match send_error.kind {
SendErrorKind::Offline => {
return (
Err(TransferError::TargetOffline(chunks_confirmed * chunk_size)),
non_transfer_message_queue,
)
}
SendErrorKind::Timeout => {
return (
Err(TransferError::TargetTimeout(chunks_confirmed * chunk_size)),
non_transfer_message_queue,
)
}
},
None => {
non_transfer_message_queue.push(next);
continue;
}
}
}
if let Ok((source, message)) = &next {
if source.process == to_addr.process {
match message {
Message::Request(_) => {
non_transfer_message_queue.push(next);
continue;
}
Message::Response((response, context)) => {
if transfer_context_id
== context
.as_ref()
.unwrap_or(&"".into())
.parse::<u64>()
.unwrap_or(0)
{
chunks_confirmed += 1;
if response
.metadata
.as_ref()
.unwrap_or(&"".into())
.parse::<u64>()
.unwrap_or(0)
!= chunks_confirmed
{
return (
Err(TransferError::TargetRejected(
chunks_confirmed * chunk_size,
)),
non_transfer_message_queue,
);
}
if chunks_confirmed == chunks_sent {
return (Ok(()), non_transfer_message_queue);
}
} else {
non_transfer_message_queue.push(next);
}
}
}
} else {
non_transfer_message_queue.push(next);
continue;
}
}
}
}
pub fn receive_transfer(
transfer_source: Address,
total_chunks: u64,
max_timeout: u64,
) -> (
Result<Vec<u8>, TransferError>,
Vec<Result<(Address, Message), (SendError, Option<Context>)>>,
) {
let start_time = std::time::SystemTime::now();
// get first payload then loop and receive rest
let mut file = match get_payload() {
Some(payload) => payload.bytes,
None => {
return (Err(TransferError::SourceFailed(0)), vec![]);
}
};
// respond to first request
send_response(
&Response {
inherit: false,
ipc: None,
metadata: Some(1.to_string()),
},
None,
);
if total_chunks == 1 {
return (Ok(file), vec![]);
}
let mut chunk_num = 1;
let mut non_transfer_message_queue = Vec::new();
loop {
let next = receive();
if start_time.elapsed().expect("time error").as_secs() > max_timeout {
return (
Err(TransferError::TargetTimeout(file.len() as u64)),
non_transfer_message_queue,
);
}
if let Err(_) = &next {
non_transfer_message_queue.push(next);
} else if let Ok((source, message)) = &next {
// we know all messages from source process will be for this transfer,
// since they are sent sequentially and it's a single-file queue.
if source.process == transfer_source.process {
match message {
Message::Request(_) => {
let payload = match get_payload() {
Some(payload) => payload,
None => {
return (
Err(TransferError::SourceFailed(file.len() as u64)),
non_transfer_message_queue,
);
}
};
chunk_num += 1;
file.extend(payload.bytes);
send_response(
&Response {
inherit: false,
ipc: None,
metadata: Some(chunk_num.to_string()),
},
None,
);
if chunk_num == total_chunks {
return (Ok(file), non_transfer_message_queue);
}
}
Message::Response(_) => {
return (
Err(TransferError::SourceFailed(file.len() as u64)),
non_transfer_message_queue,
);
}
}
} else {
non_transfer_message_queue.push(next);
continue;
}
}
}
}

5
modules/app_tracker/pkg/metadata.json
View File

@ -1,5 +0,0 @@
{
"package": "app_tracker",
"publisher": "uqbar",
"desc": "A package manager + app store. This JSON field is optional and you can add whatever you want in addition to this."
}

15
modules/app_tracker/plan.md
View File

@ -1,15 +0,0 @@
### App Tracker: our built-in package manager that lives in userspace
*note: 'app' and 'package' will be used interchangeably, but they are the same thing. generally, end users should see 'apps', and developers and the system itself should see 'packages'*
### Backend
Tracker requires full read-write to filesystem, along with caps for every other distro app and runtime module. It takes all the caps because it needs the ability to grant them to packages we install!
In order to load in the currently installed packages, Tracker will access the VFS and read from a hardcoded set of
### Frontend
Tracker will present a frontend that shows all the apps you currently have installed. You can see some metadata about them, and uninstall them from this list.
Tracker will contain a "store" to browse for new apps to install. TODO

338
modules/app_tracker/src/lib.rs
View File

@ -1,338 +0,0 @@
cargo_component_bindings::generate!();
use bindings::{
component::uq_process::types::*, get_capability, get_payload, print_to_terminal, receive, Guest, send_request, send_response
};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
#[allow(dead_code)]
mod kernel_types;
use kernel_types as kt;
#[allow(dead_code)]
mod process_lib;
struct Component;
#[derive(Debug, Serialize, Deserialize)]
pub enum AppTrackerRequest {
New { package: String },
Install { package: String },
}
#[derive(Debug, Serialize, Deserialize)]
pub enum ApptrackerResponse {
New { package: String },
Install { package: String },
Error { error: String },
}
#[derive(Debug, Serialize, Deserialize)]
pub struct PackageManifestEntry {
pub process_name: String,
pub process_wasm_path: String,
pub on_panic: kt::OnPanic,
pub request_networking: bool,
pub request_messaging: Vec<String>,
pub grant_messaging: Vec<String>, // special logic for the string "all": makes process public
}
fn parse_command(our: &Address, request_string: String) -> anyhow::Result<ApptrackerResponse> {
match serde_json::from_str(&request_string)? {
AppTrackerRequest::New { package } => {
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
let vfs_address = Address {
node: our.node.clone(),
process: ProcessId::from_str("vfs:sys:uqbar").unwrap(),
};
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(
serde_json::to_string(&kt::VfsRequest {
drive: package.clone(),
action: kt::VfsAction::New,
})
.unwrap(),
),
None,
None,
5,
)?;
// add zip bytes
let _ = process_lib::send_and_await_response(
&vfs_address,
true,
Some(
serde_json::to_string(&kt::VfsRequest {
drive: package.clone(),
action: kt::VfsAction::Add {
full_path: package.clone().into(),
entry_type: kt::AddEntryType::ZipArchive,
},
})
.unwrap(),
),
None,
Some(&payload),
5,
)?;
Ok(ApptrackerResponse::New { package })
}
AppTrackerRequest::Install { package } => {
let vfs_address = Address {
node: our.node.clone(),
process: ProcessId::from_str("vfs:sys:uqbar").unwrap(),
};
let _ = process_lib::send_and_await_response(
&vfs_address,
false,
Some(
serde_json::to_string(&kt::VfsRequest {
drive: package.clone(),
action: kt::VfsAction::GetEntry("/manifest.json".into()),
})
.unwrap(),
),
None,
None,
5,
)?;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
let manifest = String::from_utf8(payload.bytes)?;
let manifest = serde_json::from_str::<Vec<PackageManifestEntry>>(&manifest).unwrap();
for entry in manifest {
let path = if entry.process_wasm_path.starts_with("/") {
entry.process_wasm_path
} else {
format!("/{}", entry.process_wasm_path)
};
let (_, hash_response) = process_lib::send_and_await_response(
&vfs_address,
false,
Some(
serde_json::to_string(&kt::VfsRequest {
drive: package.clone(),
action: kt::VfsAction::GetHash(path.clone()),
})
.unwrap(),
),
None,
None,
5,
)?;
let Message::Response((Response { ipc: Some(ipc), .. }, _)) = hash_response else {
return Err(anyhow::anyhow!("bad vfs response"));
};
let kt::VfsResponse::GetHash(Some(hash)) = serde_json::from_str(&ipc).unwrap() else {
return Err(anyhow::anyhow!("no hash in vfs"));
};
// build initial caps
let mut initial_capabilities: HashSet<kt::SignedCapability> = HashSet::new();
if entry.request_networking {
let Some(networking_cap) = get_capability(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar").unwrap(),
},
&"\"network\"".to_string(),
) else {
return Err(anyhow::anyhow!("app_tracker: no net cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(networking_cap));
}
let Some(read_cap) = get_capability(
&vfs_address.clone(),
&serde_json::to_string(&serde_json::json!({
"kind": "read",
"drive": package,
})).unwrap(),
) else {
return Err(anyhow::anyhow!("app_tracker: no read cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(read_cap));
let Some(write_cap) = get_capability(
&vfs_address.clone(),
&serde_json::to_string(&serde_json::json!({
"kind": "write",
"drive": package,
})).unwrap(),
) else {
return Err(anyhow::anyhow!("app_tracker: no write cap"));
};
initial_capabilities.insert(kt::de_wit_signed_capability(write_cap));
let mut public = false;
for process_name in &entry.grant_messaging {
if process_name == "all" {
public = true;
continue;
}
let Ok(parsed_process_id) = ProcessId::from_str(&process_name) else {
continue;
};
let Some(messaging_cap) = get_capability(
&Address {
node: our.node.clone(),
process: parsed_process_id.clone(),
},
&"\"messaging\"".into()
) else {
return Err(anyhow::anyhow!(format!("app_tracker: no cap for {}", process_name)));
};
initial_capabilities.insert(kt::de_wit_signed_capability(messaging_cap));
}
for process_name in &entry.request_messaging {
let Ok(parsed_process_id) = ProcessId::from_str(process_name) else {
continue;
};
let Some(messaging_cap) = get_capability(
&Address {
node: our.node.clone(),
process: parsed_process_id.clone(),
},
&"\"messaging\"".into()
) else {
return Err(anyhow::anyhow!(format!("app_tracker: no cap for {}", process_name)));
};
initial_capabilities.insert(kt::de_wit_signed_capability(messaging_cap));
}
let process_id = format!("{}:{}", entry.process_name, package.clone());
let Ok(parsed_new_process_id) = ProcessId::from_str(&process_id) else {
return Err(anyhow::anyhow!("app_tracker: invalid process id!"));
};
let _ = process_lib::send_request(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar").unwrap(),
},
false,
Some(
serde_json::to_string(
&kt::KernelCommand::KillProcess(kt::ProcessId::de_wit(parsed_new_process_id.clone()))).unwrap()),
None,
None,
None,
);
// kernel start process takes bytes as payload + wasm_bytes_handle...
// reconsider perhaps
let (_, _bytes_response) = process_lib::send_and_await_response(
&vfs_address,
false,
Some(
serde_json::to_string(&kt::VfsRequest {
drive: package.clone(),
action: kt::VfsAction::GetEntry(path),
})
.unwrap(),
),
None,
None,
5,
)?;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no wasm bytes payload."));
};
let _ = process_lib::send_and_await_response(
&Address {
node: our.node.clone(),
process: ProcessId::from_str("kernel:sys:uqbar").unwrap(),
},
false,
Some(
serde_json::to_string(&kt::KernelCommand::StartProcess {
id: kt::ProcessId::de_wit(parsed_new_process_id),
wasm_bytes_handle: hash,
on_panic: entry.on_panic,
initial_capabilities,
public,
})
.unwrap(),
),
None,
Some(&payload),
5,
)?;
}
Ok(ApptrackerResponse::Install { package })
}
}
}
impl Guest for Component {
fn init(our: Address) {
assert_eq!(our.process.to_string(), "app_tracker:app_tracker:uqbar");
print_to_terminal(0, &format!("app_tracker: start"));
loop {
let message = match receive() {
Ok((source, message)) => {
if our.node != source.node {
continue;
}
message
}
Err((error, _context)) => {
print_to_terminal(0, &format!("net error: {:?}!", error.kind));
continue;
}
};
match message {
Message::Request(Request { ipc, expects_response, metadata, .. }) => {
let Some(command) = ipc else {
continue;
};
match parse_command(&our, command) {
Ok(response) => {
if let Some(_) = expects_response {
let _ = send_response(
&Response {
ipc: Some(serde_json::to_string(&response).unwrap()),
metadata,
},
None,
);
};
}
Err(e) => {
print_to_terminal(0, &format!("app_tracker: got error {}", e));
if let Some(_) = expects_response {
let error = ApptrackerResponse::Error {
error: format!("{}", e),
};
let _ = send_response(
&Response {
ipc: Some(serde_json::to_string(&error).unwrap()),
metadata,
},
None,
);
};
}
}
}
_ => continue,
}
}
}
}

2
modules/chess/pkg/manifest.json
View File

@ -9,6 +9,6 @@
"encryptor:sys:uqbar",
"http_server:sys:uqbar"
],
"grant_messaging": []
"public": false
}
]

8
modules/chess/src/lib.rs
View File

@ -85,6 +85,7 @@ fn json_game(game: &Game) -> serde_json::Value {
fn send_http_response(status: u16, headers: HashMap<String, String>, payload_bytes: Vec<u8>) {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"status": status,
@ -298,6 +299,7 @@ impl Guest for Component {
if !game.ended {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -331,6 +333,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -346,6 +349,7 @@ impl Guest for Component {
let Some(game) = state.games.get_mut(&game_id) else {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -405,6 +409,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -417,6 +422,7 @@ impl Guest for Component {
} else {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -433,6 +439,7 @@ impl Guest for Component {
let Some(game) = state.games.get_mut(&game_id) else {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -457,6 +464,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},

2
modules/homepage/pkg/manifest.json
View File

@ -9,6 +9,6 @@
"http_server:sys:uqbar",
"encryptor:sys:uqbar"
],
"grant_messaging": []
"public": false
}
]

3
modules/homepage/src/lib.rs
View File

@ -72,6 +72,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: Some(serde_json::json!({
"action": "response",
"status": 200,
@ -89,6 +90,7 @@ impl Guest for Component {
} else if message_json["path"].is_string() {
send_response(
&Response {
inherit: false,
ipc: Some(json!({
"action": "response",
"status": 404,
@ -109,6 +111,7 @@ impl Guest for Component {
} else if message_json["hello"] == "world" {
send_response(
&Response {
inherit: false,
ipc: Some(serde_json::json!({
"hello": "to you too"
}).to_string()),

4
modules/http_bindings/pkg/manifest.json
View File

@ -10,8 +10,6 @@
"encryptor:sys:uqbar",
"vfs:sys:uqbar"
],
"grant_messaging": [
"http_server:sys:uqbar"
]
"public": false
}
]

2
modules/http_bindings/src/lib.rs
View File

@ -87,6 +87,7 @@ fn auth_cookie_valid(our_node: String, cookie: &str, secret: Hmac<Sha256>) -> bo
fn send_http_response(status: u16, headers: HashMap<String, String>, payload_bytes: Vec<u8>) {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"status": status,
@ -180,6 +181,7 @@ impl Guest for Component {
};
send_response(
&Response {
inherit: false,
ipc: None,
metadata: None,
},

2
modules/http_proxy/pkg/manifest.json
View File

@ -9,6 +9,6 @@
"encryptor:sys:uqbar",
"http_server:sys:uqbar"
],
"grant_messaging": []
"public": false
}
]

6
modules/http_proxy/src/lib.rs
View File

@ -30,6 +30,7 @@ pub struct FileSystemRequest {
fn send_http_response(status: u16, headers: HashMap<String, String>, payload_bytes: Vec<u8>) {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"status": status,
@ -197,6 +198,7 @@ impl Guest for Component {
if message_json["path"] == "/http-proxy" && message_json["method"] == "GET" {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"action": "response",
@ -221,6 +223,7 @@ impl Guest for Component {
{
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"action": "response",
@ -271,6 +274,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"action": "response",
@ -313,6 +317,7 @@ impl Guest for Component {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"action": "response",
@ -349,6 +354,7 @@ impl Guest for Component {
} else if !registrations.contains_key(username) {
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"action": "response",

4
modules/key_value/pkg/manifest.json
View File

@ -7,8 +7,6 @@
"request_messaging": [
"vfs:sys:uqbar"
],
"grant_messaging": [
"all"
]
"public": true
}
]

2
modules/orgs/pkg/manifest.json
View File

@ -7,6 +7,6 @@
"request_messaging": [
"http_bindings:http_bindings:uqbar"
],
"grant_messaging": []
"public": false
}
]

4
modules/orgs/src/lib.rs
View File

@ -121,6 +121,7 @@ fn get_http_request_info(
fn send_http_response(status: u16, headers: HashMap<String, String>, payload_bytes: Vec<u8>) {
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"status": status,
@ -735,6 +736,7 @@ impl Guest for Component {
print_to_terminal(1, "orgs: get_contact_info");
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"action": "get_contact_info",
@ -758,6 +760,7 @@ impl Guest for Component {
bindings::set_state(&to_vec(&state).unwrap());
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"action": "update_contact_info",
@ -777,6 +780,7 @@ impl Guest for Component {
state.orgs.insert(org.id, org);
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"action": "update_orgs",

5
modules/qns_indexer/pkg/manifest.json
View File

@ -9,9 +9,6 @@
"http_bindings:http_bindings:uqbar",
"eth_rpc:sys:uqbar"
],
"grant_messaging": [
"eth_rpc:sys:uqbar",
"filesystem:sys:uqbar"
]
"public": true
}
]

4
modules/qns_indexer/src/lib.rs
View File

@ -201,6 +201,7 @@ impl UqProcess for Component {
if let Some(node) = state.nodes.get(name) {
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"status": 200,
@ -227,6 +228,7 @@ impl UqProcess for Component {
}
send_response(
&Response {
inherit: false,
ipc: Some(
serde_json::json!({
"status": 404,
@ -262,7 +264,7 @@ impl UqProcess for Component {
let node = &e.topics[1];
let decoded = NodeRegistered::decode_data(&decode_hex_to_vec(&e.data), true).unwrap();
let Ok(name) = dnswire_decode(decoded.0.clone()) else {
bindings::print_to_terminal(0, &format!("qns_indexer: failed to decode name: {:?}", decoded.0));
bindings::print_to_terminal(1, &format!("qns_indexer: failed to decode name: {:?}", decoded.0));
continue;
};

4
modules/rpc/pkg/manifest.json
View File

@ -6,9 +6,9 @@
"request_networking": false,
"request_messaging": [
"http_bindings:http_bindings:uqbar",
"app_tracker:app_tracker:uqbar",
"main:app_store:uqbar",
"http_server:sys:uqbar"
],
"grant_messaging": []
"public": false
}
]

2
modules/rpc/src/lib.rs
View File

@ -64,6 +64,7 @@ struct WriteFileResult {
fn send_http_response(status: u16, headers: HashMap<String, String>, payload_bytes: Vec<u8>) {
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"status": status,
@ -207,6 +208,7 @@ impl Guest for Component {
"/rpc" => {
send_response(
&Response {
inherit: false,
ipc: Some(
json!({
"action": "response",

4
modules/terminal/pkg/manifest.json
View File

@ -7,8 +7,6 @@
"request_messaging": [
"net:sys:uqbar"
],
"grant_messaging": [
"all"
]
"public": true
}
]

2
modules/terminal/src/lib.rs
View File

@ -108,7 +108,7 @@ impl Guest for Component {
};
parse_command(&our.node, command);
}
Message::Response((Response { ipc, metadata }, _)) => {
Message::Response((Response { ipc, metadata, .. }, _)) => {
if let Some(txt) = &ipc {
print_to_terminal(0, &format!("net response: {}", txt));
}

3
src/encryptor.rs
View File

@ -159,6 +159,7 @@ pub async fn encryptor(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(serde_json::json!({
"status": 201,
"headers": headers,
@ -340,6 +341,7 @@ pub async fn encryptor(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: None,
metadata: None,
},
@ -395,6 +397,7 @@ pub async fn encryptor(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: None,
metadata: None,
},

2
src/eth_rpc.rs
View File

@ -134,6 +134,7 @@ pub async fn eth_rpc(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(
serde_json::to_string::<Result<u64, EthRpcError>>(&Ok(
message.id
@ -301,6 +302,7 @@ fn make_error_message(our_name: String, km: &KernelMessage, error: EthRpcError)
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(serde_json::to_string::<Result<u64, EthRpcError>>(&Err(error)).unwrap()),
metadata: None,
},

34
src/filesystem/mod.rs
View File

@ -146,9 +146,6 @@ async fn bootstrap(
let packages: Vec<(String, zip::ZipArchive<std::io::Cursor<Vec<u8>>>)> =
get_zipped_packages().await;
// need to grant all caps at the end, after process_map has been filled in!
let mut caps_to_grant = Vec::<(ProcessId, Capability)>::new();
let mut vfs_messages = Vec::new();
for (package_name, mut package) in packages {
@ -338,26 +335,7 @@ async fn bootstrap(
.unwrap(),
});
let mut public_process = false;
// queue the granted capabilities
for process_name in &entry.grant_messaging {
if process_name == "all" {
public_process = true;
continue;
}
let process_id = ProcessId::from_str(process_name).unwrap();
caps_to_grant.push((
process_id.clone(),
Capability {
issuer: Address {
node: our_name.to_string(),
process: ProcessId::from_str(&our_process_id).unwrap(),
},
params: "\"messaging\"".into(),
},
));
}
let public_process = entry.public;
// save in process map
let file = FileIdentifier::new_uuid();
@ -376,14 +354,6 @@ async fn bootstrap(
}
}
// grant queued capabilities from all packages
for (to, cap) in caps_to_grant {
let Some(proc) = process_map.get_mut(&to) else {
continue;
};
proc.capabilities.insert(cap);
}
// save kernel process state. FsAction::SetState(kernel)
let serialized_process_map =
bincode::serialize(&process_map).expect("state map serialization error!");
@ -757,6 +727,7 @@ async fn handle_request(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(
serde_json::to_string::<Result<FsResponse, FsError>>(&Ok(ipc)).unwrap(),
),
@ -804,6 +775,7 @@ fn make_error_message(our_name: String, km: &KernelMessage, error: FsError) -> K
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(
serde_json::to_string::<Result<FsResponse, FsError>>(&Err(error)).unwrap(),
),

2
src/http_client.rs
View File

@ -144,6 +144,7 @@ async fn handle_message(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(
serde_json::to_string::<Result<HttpClientResponse, HttpClientError>>(&Ok(
http_client_response,
@ -219,6 +220,7 @@ fn make_error_message(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(
serde_json::to_string::<Result<HttpClientResponse, HttpClientError>>(&Err(
error,

1
src/http_server/server_fns.rs
View File

@ -457,6 +457,7 @@ pub fn make_error_message(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(serde_json::to_string(&error).unwrap()),
metadata: None,
},

433
src/kernel/mod.rs
View File

@ -4,12 +4,15 @@ use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet, VecDeque};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use wasmtime::component::*;
use wasmtime::{Config, Engine, Store, WasmBacktraceDetails};
use wasmtime_wasi::preview2::{DirPerms, FilePerms, Table, WasiCtx, WasiCtxBuilder, WasiView};
use wasmtime_wasi::preview2::{Table, WasiCtx, WasiCtxBuilder, WasiView};
use crate::types as t;
use crate::FILESYSTEM_PROCESS_ID;
@ -95,47 +98,47 @@ impl WasiView for ProcessWasi {
/// intercept wasi random
///
#[async_trait::async_trait]
impl wasi::random::insecure::Host for ProcessWasi {
async fn get_insecure_random_bytes(&mut self, len: u64) -> Result<Vec<u8>> {
let mut bytes = Vec::with_capacity(len as usize);
for _ in 0..len {
bytes.push(rand::random());
}
Ok(bytes)
}
// #[async_trait::async_trait]
// impl wasi::random::insecure::Host for ProcessWasi {
// async fn get_insecure_random_bytes(&mut self, len: u64) -> Result<Vec<u8>> {
// let mut bytes = Vec::with_capacity(len as usize);
// for _ in 0..len {
// bytes.push(rand::random());
// }
// Ok(bytes)
// }
async fn get_insecure_random_u64(&mut self) -> Result<u64> {
Ok(rand::random())
}
}
// async fn get_insecure_random_u64(&mut self) -> Result<u64> {
// Ok(rand::random())
// }
// }
#[async_trait::async_trait]
impl wasi::random::insecure_seed::Host for ProcessWasi {
async fn insecure_seed(&mut self) -> Result<(u64, u64)> {
Ok((rand::random(), rand::random()))
}
}
// #[async_trait::async_trait]
// impl wasi::random::insecure_seed::Host for ProcessWasi {
// async fn insecure_seed(&mut self) -> Result<(u64, u64)> {
// Ok((rand::random(), rand::random()))
// }
// }
#[async_trait::async_trait]
impl wasi::random::random::Host for ProcessWasi {
async fn get_random_bytes(&mut self, len: u64) -> Result<Vec<u8>> {
let mut bytes = Vec::with_capacity(len as usize);
getrandom::getrandom(&mut bytes[..])?;
Ok(bytes)
}
// #[async_trait::async_trait]
// impl wasi::random::random::Host for ProcessWasi {
// async fn get_random_bytes(&mut self, len: u64) -> Result<Vec<u8>> {
// let mut bytes = Vec::with_capacity(len as usize);
// getrandom::getrandom(&mut bytes[..])?;
// Ok(bytes)
// }
async fn get_random_u64(&mut self) -> Result<u64> {
let mut bytes = Vec::with_capacity(8);
getrandom::getrandom(&mut bytes[..])?;
// async fn get_random_u64(&mut self) -> Result<u64> {
// let mut bytes = Vec::with_capacity(8);
// getrandom::getrandom(&mut bytes[..])?;
let mut number = 0u64;
for (i, &byte) in bytes.iter().enumerate() {
number |= (byte as u64) << (i * 8);
}
Ok(number)
}
}
// let mut number = 0u64;
// for (i, &byte) in bytes.iter().enumerate() {
// number |= (byte as u64) << (i * 8);
// }
// Ok(number)
// }
// }
///
/// create the process API. this is where the functions that a process can use live.
@ -157,13 +160,6 @@ impl UqProcessImports for ProcessWasi {
}
}
async fn get_unix_time(&mut self) -> Result<u64> {
match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
Ok(t) => Ok(t.as_secs()),
Err(e) => Err(e.into()),
}
}
async fn get_eth_block(&mut self) -> Result<u64> {
// TODO connect to eth RPC
unimplemented!()
@ -336,6 +332,11 @@ impl UqProcessImports for ProcessWasi {
node: self.process.metadata.our.node.clone(),
process: VFS_PROCESS_ID.en_wit(),
};
let our_drive_name = [
self.process.metadata.our.process.package(),
self.process.metadata.our.process.publisher_node(),
]
.join(":");
let Ok(Ok((_, hash_response))) = send_and_await_response(
self,
None,
@ -345,7 +346,7 @@ impl UqProcessImports for ProcessWasi {
expects_response: Some(5),
ipc: Some(
serde_json::to_string(&t::VfsRequest {
drive: self.process.metadata.our.process.package().to_string(),
drive: our_drive_name.clone(),
action: t::VfsAction::GetHash(wasm_path.clone()),
})
.unwrap(),
@ -366,7 +367,6 @@ impl UqProcessImports for ProcessWasi {
let t::VfsResponse::GetHash(Some(hash)) = serde_json::from_str(&ipc).unwrap() else {
return Ok(Err(wit::SpawnError::NoFileAtPath));
};
let Ok(Ok(_)) = send_and_await_response(
self,
None,
@ -376,7 +376,7 @@ impl UqProcessImports for ProcessWasi {
expects_response: Some(5),
ipc: Some(
serde_json::to_string(&t::VfsRequest {
drive: self.process.metadata.our.process.package().to_string(),
drive: our_drive_name,
action: t::VfsAction::GetEntry(wasm_path.clone()),
})
.unwrap(),
@ -389,7 +389,6 @@ impl UqProcessImports for ProcessWasi {
else {
return Ok(Err(wit::SpawnError::NoFileAtPath));
};
let Some(t::Payload { mime: _, ref bytes }) = self.process.last_payload else {
return Ok(Err(wit::SpawnError::NoFileAtPath));
};
@ -403,7 +402,6 @@ impl UqProcessImports for ProcessWasi {
self.process.metadata.our.process.package(),
self.process.metadata.our.process.publisher_node(),
);
let Ok(Ok((_, response))) = send_and_await_response(
self,
Some(t::Address {
@ -474,14 +472,12 @@ impl UqProcessImports for ProcessWasi {
else {
return Ok(Err(wit::SpawnError::NameTaken));
};
let wit::Message::Response((wit::Response { ipc: Some(ipc), .. }, _)) = response else {
return Ok(Err(wit::SpawnError::NoFileAtPath));
};
let t::KernelResponse::StartedProcess = serde_json::from_str(&ipc).unwrap() else {
return Ok(Err(wit::SpawnError::NoFileAtPath));
};
// child processes are always able to Message parent
let (tx, rx) = tokio::sync::oneshot::channel();
self.process
@ -516,7 +512,6 @@ impl UqProcessImports for ProcessWasi {
.await
.unwrap();
let _ = rx.await.unwrap();
Ok(Ok(new_process_id.en_wit().to_owned()))
}
@ -619,9 +614,8 @@ impl UqProcessImports for ProcessWasi {
cap: cap.clone(),
responder: tx,
})
.await
.unwrap();
let _ = rx.await.unwrap();
.await?;
let _ = rx.await?;
}
Ok(())
}
@ -678,8 +672,36 @@ impl UqProcessImports for ProcessWasi {
},
responder: tx,
})
.await;
let _ = rx.await.unwrap();
.await?;
let _ = rx.await?;
Ok(())
}
async fn share_capability(
&mut self,
to: wit::ProcessId,
signed_cap: wit::SignedCapability,
) -> Result<()> {
let pk = signature::UnparsedPublicKey::new(
&signature::ED25519,
self.process.keypair.public_key(),
);
let cap = t::Capability {
issuer: t::Address::de_wit(signed_cap.issuer),
params: signed_cap.params,
};
pk.verify(&bincode::serialize(&cap).unwrap(), &signed_cap.signature)?;
let (tx, rx) = tokio::sync::oneshot::channel();
let _ = self
.process
.caps_oracle
.send(t::CapMessage::Add {
on: t::ProcessId::de_wit(to),
cap,
responder: tx,
})
.await?;
let _ = rx.await?;
Ok(())
}
//
@ -966,7 +988,10 @@ impl Process {
mime: p.mime,
bytes: p.bytes,
}),
None => None,
None => match request.inherit {
true => self.last_payload.clone(),
false => None,
},
};
// rsvp is set if there was a Request expecting Response
@ -1007,7 +1032,7 @@ impl Process {
let self_sender = self.self_sender.clone();
let timeout_handle = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await;
self_sender
let _ = self_sender
.send(Err(t::WrappedSendError {
id: request_id,
source: t::Address::de_wit(target.clone()), // TODO check this
@ -1018,8 +1043,7 @@ impl Process {
payload,
},
}))
.await
.unwrap();
.await;
});
self.save_context(kernel_message.id, new_context, timeout_handle)
.await;
@ -1049,6 +1073,11 @@ impl Process {
}
};
let payload = match response.inherit {
true => self.last_payload.clone(),
false => de_wit_payload(payload),
};
self.send_to_loop
.send(t::KernelMessage {
id,
@ -1060,7 +1089,7 @@ impl Process {
// the context will be set by the process receiving this Response.
None,
)),
payload: de_wit_payload(payload),
payload,
signed_capabilities: None,
})
.await
@ -1106,20 +1135,44 @@ async fn persist_state(
/// create a specific process, and generate a task that will run it.
async fn make_process_loop(
booted: Arc<AtomicBool>,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
metadata: t::ProcessMetadata,
send_to_loop: t::MessageSender,
send_to_terminal: t::PrintSender,
recv_in_process: ProcessMessageReceiver,
mut recv_in_process: ProcessMessageReceiver,
send_to_process: ProcessMessageSender,
wasm_bytes: &Vec<u8>,
caps_oracle: t::CapMessageSender,
engine: &Engine,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
// let dir = std::env::current_dir().unwrap();
let dir = cap_std::fs::Dir::open_ambient_dir(home_directory_path, cap_std::ambient_authority())
.unwrap();
// before process can be instantiated, need to await booted message from kernel
if !booted.load(Ordering::Relaxed) {
let mut pre_boot_queue = Vec::<Result<t::KernelMessage, t::WrappedSendError>>::new();
while let Some(message) = recv_in_process.recv().await {
if let Err(_) = &message {
pre_boot_queue.push(message);
continue;
}
let message = message.unwrap();
if (message.source
== t::Address {
node: metadata.our.node.clone(),
process: KERNEL_PROCESS_ID.clone(),
})
&& (message.message
== t::Message::Request(t::Request {
inherit: false,
expects_response: None,
ipc: Some("booted".into()),
metadata: None,
}))
{
break;
}
pre_boot_queue.push(Ok(message));
}
}
let component =
Component::new(&engine, wasm_bytes).expect("make_process_loop: couldn't read file");
@ -1128,30 +1181,28 @@ async fn make_process_loop(
UqProcess::add_to_linker(&mut linker, |state: &mut ProcessWasi| state).unwrap();
let mut table = Table::new();
let wasi = WasiCtxBuilder::new()
.push_preopened_dir(dir, DirPerms::all(), FilePerms::all(), &"")
.build(&mut table)
.unwrap();
let wasi = WasiCtxBuilder::new().build(&mut table).unwrap();
// wasmtime_wasi::preview2::command::add_to_linker(&mut linker).unwrap();
wasmtime_wasi::preview2::bindings::clocks::wall_clock::add_to_linker(&mut linker, |t| t)
.unwrap();
wasmtime_wasi::preview2::bindings::clocks::monotonic_clock::add_to_linker(&mut linker, |t| t)
.unwrap();
wasmtime_wasi::preview2::bindings::clocks::timezone::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::filesystem::filesystem::add_to_linker(&mut linker, |t| t)
.unwrap();
wasmtime_wasi::preview2::bindings::poll::poll::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::io::streams::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::command::add_to_linker(&mut linker).unwrap();
// wasmtime_wasi::preview2::bindings::clocks::wall_clock::add_to_linker(&mut linker, |t| t)
// .unwrap();
// wasmtime_wasi::preview2::bindings::clocks::monotonic_clock::add_to_linker(&mut linker, |t| t)
// .unwrap();
// wasmtime_wasi::preview2::bindings::clocks::timezone::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::filesystem::filesystem::add_to_linker(&mut linker, |t| t)
// .unwrap();
// wasmtime_wasi::preview2::bindings::poll::poll::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::io::streams::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::random::random::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::cli_base::exit::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::cli_base::environment::add_to_linker(&mut linker, |t| t)
.unwrap();
wasmtime_wasi::preview2::bindings::cli_base::preopens::add_to_linker(&mut linker, |t| t)
.unwrap();
wasmtime_wasi::preview2::bindings::cli_base::stdin::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::cli_base::stdout::add_to_linker(&mut linker, |t| t).unwrap();
wasmtime_wasi::preview2::bindings::cli_base::stderr::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::exit::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::environment::add_to_linker(&mut linker, |t| t)
// .unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::preopens::add_to_linker(&mut linker, |t| t)
// .unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::stdin::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::stdout::add_to_linker(&mut linker, |t| t).unwrap();
// wasmtime_wasi::preview2::bindings::cli_base::stderr::add_to_linker(&mut linker, |t| t).unwrap();
let mut store = Store::new(
engine,
ProcessWasi {
@ -1194,15 +1245,24 @@ async fn make_process_loop(
// the process will run until it returns from init()
let is_error = match bindings.call_init(&mut store, &metadata.our.en_wit()).await {
Ok(()) => false,
Ok(()) => {
let _ =
send_to_terminal
.send(t::Printout {
verbosity: 1,
content: format!(
"process {} returned without error",
metadata.our.process,
),
})
.await;
false
}
Err(e) => {
let _ = send_to_terminal
.send(t::Printout {
verbosity: 0,
content: format!(
"mk: process {:?} ended with error:",
metadata.our.process,
),
content: format!("process {:?} ended with error:", metadata.our.process,),
})
.await;
for line in format!("{:?}", e).lines() {
@ -1321,8 +1381,8 @@ async fn make_process_loop(
/// handle messages sent directly to kernel. source is always our own node.
async fn handle_kernel_request(
our_name: String,
booted: Arc<AtomicBool>,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
km: t::KernelMessage,
send_to_loop: t::MessageSender,
send_to_terminal: t::PrintSender,
@ -1349,6 +1409,36 @@ async fn handle_kernel_request(
Ok(c) => c,
};
match command {
t::KernelCommand::Booted => {
for (process_id, process_sender) in senders {
let ProcessSender::Userspace(sender) = process_sender else {
continue;
};
let _ = sender
.send(Ok(t::KernelMessage {
id: rand::random(),
source: t::Address {
node: our_name.clone(),
process: KERNEL_PROCESS_ID.clone(),
},
target: t::Address {
node: our_name.clone(),
process: process_id.clone(),
},
rsvp: None,
message: t::Message::Request(t::Request {
inherit: false,
expects_response: None,
ipc: Some("booted".into()),
metadata: None,
}),
payload: None,
signed_capabilities: None,
}))
.await;
}
booted.store(true, Ordering::Relaxed);
}
t::KernelCommand::Shutdown => {
for handle in process_handles.values() {
handle.abort();
@ -1384,6 +1474,7 @@ async fn handle_kernel_request(
rsvp: None,
message: t::Message::Response((
t::Response {
inherit: false,
ipc: Some(
serde_json::to_string(&t::KernelResponse::StartProcessError)
.unwrap(),
@ -1419,11 +1510,20 @@ async fn handle_kernel_request(
valid_capabilities.insert(cap);
}
// always give process the messaging cap for itself
valid_capabilities.insert(t::Capability {
issuer: t::Address {
node: our_name.clone(),
process: id.clone(),
},
params: "\"messaging\"".into(),
});
// fires "success" response back
start_process(
our_name,
booted,
keypair.clone(),
home_directory_path,
km.id,
&payload.bytes,
send_to_loop,
@ -1540,6 +1640,7 @@ async fn handle_kernel_request(
rsvp: None,
message: t::Message::Response((
t::Response {
inherit: false,
ipc: Some(
serde_json::to_string(&t::KernelResponse::KilledProcess(
process_id,
@ -1566,8 +1667,8 @@ async fn handle_kernel_request(
// `let meta: StartProcessMetadata ... `
async fn handle_kernel_response(
our_name: String,
booted: Arc<AtomicBool>,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
km: t::KernelMessage,
send_to_loop: t::MessageSender,
send_to_terminal: t::PrintSender,
@ -1628,8 +1729,8 @@ async fn handle_kernel_response(
start_process(
our_name,
booted,
keypair.clone(),
home_directory_path,
km.id,
&payload.bytes,
send_to_loop,
@ -1646,8 +1747,8 @@ async fn handle_kernel_response(
async fn start_process(
our_name: String,
booted: Arc<AtomicBool>,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
km_id: u64,
km_payload_bytes: &Vec<u8>,
send_to_loop: t::MessageSender,
@ -1701,8 +1802,8 @@ async fn start_process(
process_id.clone(),
tokio::spawn(
make_process_loop(
booted,
keypair.clone(),
home_directory_path,
metadata.clone(),
send_to_loop.clone(),
send_to_terminal.clone(),
@ -1734,6 +1835,7 @@ async fn start_process(
rsvp: None,
message: t::Message::Response((
t::Response {
inherit: false,
ipc: Some(serde_json::to_string(&t::KernelResponse::StartedProcess).unwrap()),
metadata: None,
},
@ -1751,7 +1853,6 @@ async fn start_process(
async fn make_event_loop(
our_name: String,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
mut process_map: t::ProcessMap,
caps_oracle_sender: t::CapMessageSender,
mut caps_oracle_receiver: t::CapMessageReceiver,
@ -1769,6 +1870,9 @@ async fn make_event_loop(
send_to_terminal: t::PrintSender,
engine: Engine,
) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
// shared global flag to mark if we're finished boot process
let booted = Arc::new(AtomicBool::new(false));
Box::pin(async move {
let mut senders: Senders = HashMap::new();
senders.insert(
@ -1839,6 +1943,7 @@ async fn make_event_loop(
}
if let t::OnPanic::Requests(requests) = &persisted.on_panic {
// if a persisted process had on-death-requests, we should perform them now
// TODO check for caps here
for (address, request, payload) in requests {
// the process that made the request is dead, so never expects response
let mut request = request.clone();
@ -1862,10 +1967,35 @@ async fn make_event_loop(
}
}
// after all bootstrapping messages are handled, send a Booted kernelcommand
// to turn it on
send_to_loop
.send(t::KernelMessage {
id: rand::random(),
source: t::Address {
node: our_name.clone(),
process: KERNEL_PROCESS_ID.clone(),
},
target: t::Address {
node: our_name.clone(),
process: KERNEL_PROCESS_ID.clone(),
},
rsvp: None,
message: t::Message::Request(t::Request {
inherit: true,
expects_response: None,
ipc: Some(serde_json::to_string(&t::KernelCommand::Booted).unwrap()),
metadata: None,
}),
payload: None,
signed_capabilities: None,
})
.await
.unwrap();
// main message loop
loop {
tokio::select! {
// aaa
// debug mode toggle: when on, this loop becomes a manual step-through
debug = recv_debug_in_loop.recv() => {
if let Some(t::DebugCommand::Toggle) = debug {
@ -1912,7 +2042,8 @@ async fn make_event_loop(
//
// enforce capabilities by matching from our set based on fixed format
// enforce that if message is directed over the network, process has capability to do so
if kernel_message.target.node != our_name {
if kernel_message.source.node == our_name
&& kernel_message.target.node != our_name {
if !process_map.get(&kernel_message.source.process).unwrap().capabilities.contains(
&t::Capability {
issuer: t::Address {
@ -1927,50 +2058,74 @@ async fn make_event_loop(
t::Printout {
verbosity: 0,
content: format!(
"event loop: process {:?} doesn't have capability to send networked messages",
"event loop: process {} doesn't have capability to send networked messages",
kernel_message.source.process
)
}
).await;
continue;
}
} else if kernel_message.source.node != our_name {
// note that messaging restrictions only apply to *local* processes, if your
// process has networking capabilities, it can be messaged by any process remotely..
let Some(persisted) = process_map.get(&kernel_message.target.process) else {
println!("kernel: did not find process in process_map: {}\r", kernel_message.target.process);
continue;
};
if !persisted.capabilities.contains(
&t::Capability {
issuer: t::Address {
node: our_name.clone(),
process: KERNEL_PROCESS_ID.clone(),
},
params: "\"network\"".into(),
}) {
// capabilities are not correct! skip this message.
let _ = send_to_terminal.send(
t::Printout {
verbosity: 0,
content: format!(
"event loop: process {} doesn't have capability to receive networked messages",
kernel_message.target.process
)
}
).await;
continue;
}
} else {
// enforce that process has capability to message a target process of this name
// kernel and filesystem can ALWAYS message any process
// enforce that local process has capability to message a target process of this name
// kernel and filesystem can ALWAYS message any local process
if kernel_message.source.process != *KERNEL_PROCESS_ID
&& kernel_message.source.process != *FILESYSTEM_PROCESS_ID
{
let is_target_public = match process_map.get(&kernel_message.target.process) {
None => false,
Some(p) => p.public,
let Some(persisted_source) = process_map.get(&kernel_message.source.process) else {
println!("kernel: did not find process in process_map: {}\r", kernel_message.source.process);
continue;
};
if !is_target_public {
match process_map.get(&kernel_message.source.process) {
None => {
println!("kernel: did not find process in process_map: {}\r", kernel_message.source.process);
}, // this should only get hit by kernel?
Some(persisted) => {
if !persisted.capabilities.contains(&t::Capability {
issuer: t::Address {
node: our_name.clone(),
process: kernel_message.target.process.clone(),
},
params: "\"messaging\"".into(),
}) {
// capabilities are not correct! skip this message.
// TODO some kind of error thrown back at process
let _ = send_to_terminal.send(
t::Printout {
verbosity: 0,
content: format!(
"event loop: process {:?} doesn't have capability to message process {:?}",
kernel_message.source.process, kernel_message.target.process
)
}
).await;
continue;
let Some(persisted_target) = process_map.get(&kernel_message.target.process) else {
println!("kernel: did not find process in process_map: {}\r", kernel_message.target.process);
continue;
};
if !persisted_target.public {
if !persisted_source.capabilities.contains(&t::Capability {
issuer: t::Address {
node: our_name.clone(),
process: kernel_message.target.process.clone(),
},
params: "\"messaging\"".into(),
}) {
// capabilities are not correct! skip this message.
// TODO some kind of error thrown back at process
let _ = send_to_terminal.send(
t::Printout {
verbosity: 0,
content: format!(
"event loop: process {} doesn't have capability to message process {}",
kernel_message.source.process, kernel_message.target.process
)
}
}
).await;
continue;
}
}
}
@ -2002,8 +2157,8 @@ async fn make_event_loop(
t::Message::Request(_) => {
handle_kernel_request(
our_name.clone(),
booted.clone(),
keypair.clone(),
home_directory_path.clone(),
kernel_message,
send_to_loop.clone(),
send_to_terminal.clone(),
@ -2017,8 +2172,8 @@ async fn make_event_loop(
t::Message::Response(_) => {
handle_kernel_response(
our_name.clone(),
booted.clone(),
keypair.clone(),
home_directory_path.clone(),
kernel_message,
send_to_loop.clone(),
send_to_terminal.clone(),
@ -2129,7 +2284,6 @@ async fn make_event_loop(
pub async fn kernel(
our: t::Identity,
keypair: Arc<signature::Ed25519KeyPair>,
home_directory_path: String,
process_map: t::ProcessMap,
caps_oracle_sender: t::CapMessageSender,
caps_oracle_receiver: t::CapMessageReceiver,
@ -2157,7 +2311,6 @@ pub async fn kernel(
make_event_loop(
our.name,
keypair,
home_directory_path,
process_map,
caps_oracle_sender,
caps_oracle_receiver,

2
src/kernel/utils.rs
View File

@ -34,6 +34,7 @@ pub fn en_wit_request(request: t::Request) -> wit::Request {
pub fn de_wit_response(wit: wit::Response) -> t::Response {
t::Response {
inherit: wit.inherit,
ipc: wit.ipc,
metadata: wit.metadata,
}
@ -41,6 +42,7 @@ pub fn de_wit_response(wit: wit::Response) -> t::Response {
pub fn en_wit_response(response: t::Response) -> wit::Response {
wit::Response {
inherit: response.inherit,
ipc: response.ipc,
metadata: response.metadata,
}

3
src/kernel_types.rs
View File

@ -137,6 +137,7 @@ pub struct Request {
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Response {
pub inherit: bool,
pub ipc: Option<String>, // JSON-string
pub metadata: Option<String>, // JSON-string
}
@ -334,6 +335,7 @@ pub fn en_wit_request(request: Request) -> wit::Request {
pub fn de_wit_response(wit: wit::Response) -> Response {
Response {
inherit: wit.inherit,
ipc: wit.ipc,
metadata: wit.metadata,
}
@ -341,6 +343,7 @@ pub fn de_wit_response(wit: wit::Response) -> Response {
pub fn en_wit_response(response: Response) -> wit::Response {
wit::Response {
inherit: response.inherit,
ipc: response.ipc,
metadata: response.metadata,
}

1
src/main.rs
View File

@ -245,7 +245,6 @@ async fn main() {
tasks.spawn(kernel::kernel(
our.clone(),
networking_keypair_arc.clone(),
home_directory_path.into(),
kernel_process_map.clone(),
caps_oracle_sender.clone(),
caps_oracle_receiver,

5
src/net/mod.rs
View File

@ -718,6 +718,7 @@ async fn handle_incoming_message(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some("delivered".into()),
metadata: None,
},
@ -781,7 +782,7 @@ async fn handle_incoming_message(
NetActions::QnsUpdate(log) => {
let _ = print_tx
.send(Printout {
verbosity: 0, // TODO 1
verbosity: 1,
content: format!("net: got QNS update for {}", log.name),
})
.await;
@ -804,7 +805,7 @@ async fn handle_incoming_message(
NetActions::QnsBatchUpdate(log_list) => {
let _ = print_tx
.send(Printout {
verbosity: 0, // TODO 1
verbosity: 1,
content: format!(
"net: got QNS update with {} peers",
log_list.len()

136
src/process_lib.rs
View File

@ -3,6 +3,49 @@ use serde::{Deserialize, Serialize};
use super::bindings::component::uq_process::types::*;
use super::bindings::{Address, Payload, ProcessId, SendError};
#[derive(Hash, Eq, PartialEq, Debug, Clone, Serialize, Deserialize)]
pub struct PackageId {
pub package_name: String,
pub publisher_node: String,
}
impl PackageId {
pub fn new(package_name: &str, publisher_node: &str) -> Self {
PackageId {
package_name: package_name.into(),
publisher_node: publisher_node.into(),
}
}
pub fn from_str(input: &str) -> Result<Self, ProcessIdParseError> {
// split string on colons into 2 segments
let mut segments = input.split(':');
let package_name = segments
.next()
.ok_or(ProcessIdParseError::MissingField)?
.to_string();
let publisher_node = segments
.next()
.ok_or(ProcessIdParseError::MissingField)?
.to_string();
if segments.next().is_some() {
return Err(ProcessIdParseError::TooManyColons);
}
Ok(PackageId {
package_name,
publisher_node,
})
}
pub fn to_string(&self) -> String {
[self.package_name.as_str(), self.publisher_node.as_str()].join(":")
}
pub fn package(&self) -> &str {
&self.package_name
}
pub fn publisher_node(&self) -> &str {
&self.publisher_node
}
}
#[allow(dead_code)]
impl ProcessId {
/// generates a random u64 number if process_name is not declared
@ -56,6 +99,16 @@ impl ProcessId {
}
}
impl std::fmt::Display for ProcessId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{}:{}:{}",
self.process_name, self.package_name, self.publisher_node
)
}
}
impl PartialEq for ProcessId {
fn eq(&self, other: &Self) -> bool {
self.process_name == other.process_name
@ -64,12 +117,95 @@ impl PartialEq for ProcessId {
}
}
impl PartialEq<&str> for ProcessId {
fn eq(&self, other: &&str) -> bool {
&self.to_string() == other
}
}
impl PartialEq<ProcessId> for &str {
fn eq(&self, other: &ProcessId) -> bool {
self == &other.to_string()
}
}
#[derive(Debug)]
pub enum ProcessIdParseError {
TooManyColons,
MissingField,
}
impl std::fmt::Display for ProcessIdParseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{}",
match self {
ProcessIdParseError::TooManyColons => "Too many colons in ProcessId string",
ProcessIdParseError::MissingField => "Missing field in ProcessId string",
}
)
}
}
impl std::error::Error for ProcessIdParseError {
fn description(&self) -> &str {
match self {
ProcessIdParseError::TooManyColons => "Too many colons in ProcessId string",
ProcessIdParseError::MissingField => "Missing field in ProcessId string",
}
}
}
impl Address {
pub fn from_str(input: &str) -> Result<Self, AddressParseError> {
// split string on colons into 4 segments,
// first one with @, next 3 with :
let mut name_rest = input.split('@');
let node = name_rest
.next()
.ok_or(AddressParseError::MissingField)?
.to_string();
let mut segments = name_rest
.next()
.ok_or(AddressParseError::MissingNodeId)?
.split(':');
let process_name = segments
.next()
.ok_or(AddressParseError::MissingField)?
.to_string();
let package_name = segments
.next()
.ok_or(AddressParseError::MissingField)?
.to_string();
let publisher_node = segments
.next()
.ok_or(AddressParseError::MissingField)?
.to_string();
if segments.next().is_some() {
return Err(AddressParseError::TooManyColons);
}
Ok(Address {
node,
process: ProcessId {
process_name,
package_name,
publisher_node,
},
})
}
pub fn to_string(&self) -> String {
[self.node.as_str(), &self.process.to_string()].join("@")
}
}
#[derive(Debug)]
pub enum AddressParseError {
TooManyColons,
MissingNodeId,
MissingField,
}
pub fn send_and_await_response(
target: &Address,
inherit: bool,

17
src/types.rs
View File

@ -228,7 +228,7 @@ pub struct Payload {
pub bytes: Vec<u8>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Request {
pub inherit: bool,
pub expects_response: Option<u64>, // number of seconds until timeout
@ -236,13 +236,14 @@ pub struct Request {
pub metadata: Option<String>, // JSON-string
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Response {
pub inherit: bool,
pub ipc: Option<String>, // JSON-string
pub metadata: Option<String>, // JSON-string
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum Message {
Request(Request),
Response((Response, Option<Context>)),
@ -355,6 +356,7 @@ pub enum DebugCommand {
#[derive(Debug, Serialize, Deserialize)]
pub enum KernelCommand {
Booted,
StartProcess {
id: ProcessId,
wasm_bytes_handle: u128,
@ -439,7 +441,7 @@ pub struct PackageManifestEntry {
pub on_panic: OnPanic,
pub request_networking: bool,
pub request_messaging: Vec<String>,
pub grant_messaging: Vec<String>, // special logic for the string "all": makes process public
pub public: bool,
}
#[derive(Serialize, Deserialize, Debug)]
@ -703,8 +705,8 @@ impl std::fmt::Display for KernelMessage {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(
f,
"{{\n id: {},\n source: {},\n target: {},\n rsvp: {:?},\n message: {}\n}}",
self.id, self.source, self.target, self.rsvp, self.message,
"{{\n id: {},\n source: {},\n target: {},\n rsvp: {:?},\n message: {},\n payload: {}\n}}",
self.id, self.source, self.target, self.rsvp, self.message, self.payload.is_some()
)
}
}
@ -722,7 +724,8 @@ impl std::fmt::Display for Message {
),
Message::Response((response, context)) => write!(
f,
"Response(\n ipc: {},\n metadata: {},\n context: {}\n )",
"Response(\n inherit: {},\n ipc: {},\n metadata: {},\n context: {}\n )",
response.inherit,
&response.ipc.as_ref().unwrap_or(&"None".into()),
&response.metadata.as_ref().unwrap_or(&"None".into()),
&context.as_ref().unwrap_or(&"None".into()),

24
src/vfs.rs
View File

@ -113,6 +113,7 @@ fn make_error_message(
rsvp: None,
message: Message::Response((
Response {
inherit: false,
ipc: Some(serde_json::to_string(&VfsResponse::Err(error)).unwrap()), // TODO: handle error?
metadata: None,
},
@ -638,7 +639,14 @@ async fn handle_request(
process: source.process.clone(),
},
rsvp,
message: Message::Response((Response { ipc, metadata }, None)),
message: Message::Response((
Response {
inherit: false,
ipc,
metadata,
},
None,
)),
payload: match bytes {
Some(bytes) => Some(Payload {
mime: Some("application/octet-stream".into()),
@ -986,8 +994,7 @@ async fn match_request(
}
};
let KernelMessage { message, .. } = write_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message
else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("")
};
let Some(ipc) = ipc else {
@ -1220,7 +1227,7 @@ async fn match_request(
.await;
let write_response = recv_response.recv().await.unwrap();
let KernelMessage { message, .. } = write_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("")
};
let Some(ipc) = ipc else {
@ -1275,7 +1282,7 @@ async fn match_request(
.await;
let write_response = recv_response.recv().await.unwrap();
let KernelMessage { message, .. } = write_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("")
};
let Some(ipc) = ipc else {
@ -1419,8 +1426,7 @@ async fn match_request(
let KernelMessage {
message, payload, ..
} = read_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message
else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("");
};
let Some(ipc) = ipc else {
@ -1503,7 +1509,7 @@ async fn match_request(
let KernelMessage {
message, payload, ..
} = read_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("")
};
let Some(ipc) = ipc else {
@ -1568,7 +1574,7 @@ async fn match_request(
.await;
let length_response = recv_response.recv().await.unwrap();
let KernelMessage { message, .. } = length_response;
let Message::Response((Response { ipc, metadata: _ }, None)) = message else {
let Message::Response((Response { ipc, .. }, None)) = message else {
panic!("")
};
let Some(ipc) = ipc else {

26
wit/deps/random/insecure-seed.wit
View File

@ -1,26 +0,0 @@
// https://github.com/bytecodealliance/wasmtime/blob/432b5471ec4bf6d51173def284cd418be6849a49/crates/wasi/wit/deps/random/insecure-seed.wit
/// The insecure-seed interface for seeding hash-map DoS resistance.
///
/// It is intended to be portable at least between Unix-family platforms and
/// Windows.
interface insecure-seed {
/// Return a 128-bit value that may contain a pseudo-random value.
///
/// The returned value is not required to be computed from a CSPRNG, and may
/// even be entirely deterministic. Host implementations are encouraged to
/// provide pseudo-random values to any program exposed to
/// attacker-controlled content, to enable DoS protection built into many
/// languages' hash-map implementations.
///
/// This function is intended to only be called once, by a source language
/// to initialize Denial Of Service (DoS) protection in its hash-map
/// implementation.
///
/// # Expected future evolution
///
/// This will likely be changed to a value import, to prevent it from being
/// called multiple times and potentially used for purposes other than DoS
/// protection.
insecure-seed: func() -> tuple<u64, u64>
}

23
wit/deps/random/insecure.wit
View File

@ -1,23 +0,0 @@
// https://github.com/bytecodealliance/wasmtime/blob/432b5471ec4bf6d51173def284cd418be6849a49/crates/wasi/wit/deps/random/insecure.wit
/// The insecure interface for insecure pseudo-random numbers.
///
/// It is intended to be portable at least between Unix-family platforms and
/// Windows.
interface insecure {
/// Return `len` insecure pseudo-random bytes.
///
/// This function is not cryptographically secure. Do not use it for
/// anything related to security.
///
/// There are no requirements on the values of the returned bytes, however
/// implementations are encouraged to return evenly distributed values with
/// a long period.
get-insecure-random-bytes: func(len: u64) -> list<u8>
/// Return an insecure pseudo-random `u64` value.
///
/// This function returns the same type of pseudo-random data as
/// `get-insecure-random-bytes`, represented as a `u64`.
get-insecure-random-u64: func() -> u64
}

23
wit/deps/random/random.wit
View File

@ -1,23 +0,0 @@
// https://github.com/bytecodealliance/wasmtime/blob/432b5471ec4bf6d51173def284cd418be6849a49/crates/wasi/wit/deps/random/random.wit
package wasi:random
interface random {
/// Return `len` cryptographically-secure pseudo-random bytes.
///
/// This function must produce data from an adequately seeded
/// cryptographically-secure pseudo-random number generator (CSPRNG), so it
/// must not block, from the perspective of the calling program, and the
/// returned data is always unpredictable.
///
/// This function must always return fresh pseudo-random data. Deterministic
/// environments must omit this function, rather than implementing it with
/// deterministic data.
get-random-bytes: func(len: u64) -> list<u8>
/// Return a cryptographically-secure pseudo-random `u64` value.
///
/// This function returns the same type of pseudo-random data as
/// `get-random-bytes`, represented as a `u64`.
get-random-u64: func() -> u64
}

14
wit/uqbar.wit
View File

@ -1,4 +1,4 @@
package component:uq-process
package component:uq-process@0.1.0
interface types {
// JSON is passed over WASM boundary as a string.
@ -41,6 +41,7 @@ interface types {
}
record response {
inherit: bool,
ipc: option<json>,
metadata: option<json>,
// to grab payload, use get_payload()
@ -129,7 +130,8 @@ world uq-process {
// system utils:
import print-to-terminal: func(verbosity: u8, message: string)
import get-unix-time: func() -> u64
// **more will be added here with regard to blockchains**
import get-eth-block: func() -> u64
// process management:
@ -172,6 +174,9 @@ world uq-process {
// which must be a locally-running process.
import create-capability: func(to: process-id, params: json)
// take a signed capability and save it to a given locally-running process
import share-capability: func(to: process-id, capability: signed-capability)
// message I/O:
@ -195,9 +200,4 @@ world uq-process {
import send-and-await-response:
func(target: address, request: request, payload: option<payload>) ->
result<tuple<address, message>, send-error>
// wasi
import wasi:random/insecure
import wasi:random/insecure-seed
import wasi:random/random
}