uqbar-dao/nectar
1
1
Fork 0
mirror of https://github.com/uqbar-dao/nectar.git synced 2024-12-20 07:01:40 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #117 from uqbar-dao/bp/backup

Browse Source 
backup
This commit is contained in:
dr-frmr 2023-12-27 01:38:28 -05:00 committed by GitHub
commit 6d12014c8e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 888 additions and 797 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
modules/app_store/app_store/Cargo.lock generated
View File

@ -206,6 +206,22 @@ version = "0.4.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f"
[[package]]
name = "mime"
version = "0.3.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
[[package]]
name = "mime_guess"
version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4192263c238a5f0d0c6bfd21f336a313a4ce1c450542449ca191bb657b4642ef"
dependencies = [
"mime",
"unicase",
]
[[package]]
name = "percent-encoding"
version = "2.3.1"
@ -387,6 +403,15 @@ version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
[[package]]
name = "unicase"
version = "2.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7d2d4dafb69621809a81864c9c1b864479e1235c0dd4e199924b9742439ed89"
dependencies = [
"version_check",
]
[[package]]
name = "unicode-bidi"
version = "0.3.14"
@ -423,11 +448,12 @@ checksum = "f962df74c8c05a667b5ee8bcf162993134c104e96440b663c8daa176dc772d8c"
[[package]]
name = "uqbar_process_lib"
version = "0.4.0"
source = "git+ssh://git@github.com/uqbar-dao/process_lib.git?rev=8342b1a#8342b1a131401fb5d141dab8c90e79aa6d2bc909"
source = "git+ssh://git@github.com/uqbar-dao/process_lib.git?rev=2d17d75#2d17d75152e55ef3ed417c79312e209ca45b8dbb"
dependencies = [
"anyhow",
"bincode",
"http",
"mime_guess",
"rand",
"serde",
"serde_json",

4
modules/app_store/app_store/Cargo.toml
View File

@ -3,8 +3,6 @@ name = "app_store"
version = "0.2.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[profile.release]
panic = "abort"
opt-level = "s"
@ -17,7 +15,7 @@ rand = "0.8"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
sha2 = "0.10.8"
uqbar_process_lib = { git = "ssh://git@github.com/uqbar-dao/process_lib.git", rev = "8342b1a" }
uqbar_process_lib = { git = "ssh://git@github.com/uqbar-dao/process_lib.git", rev = "2d17d75" }
wit-bindgen = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "efcc759" }
[lib]

467
modules/app_store/app_store/src/lib.rs
View File

@ -1,11 +1,10 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::Digest;
use std::collections::{HashMap, HashSet};
use uqbar_process_lib::kernel_types as kt;
use uqbar_process_lib::{
await_message, get_capability, get_payload, get_typed_state, grant_messaging, println,
set_state, share_capability, Address, Message, NodeId, PackageId, ProcessId, Request, Response,
};
use uqbar_process_lib::println;
use uqbar_process_lib::*;
wit_bindgen::generate!({
path: "../../../wit",
@ -15,14 +14,11 @@ wit_bindgen::generate!({
},
});
#[allow(dead_code)]
mod ft_worker_lib;
use ft_worker_lib::{
spawn_receive_transfer, spawn_transfer, FTWorkerCommand, FTWorkerResult, FileTransferContext,
};
struct Component;
/// Uqbar App Store:
/// acts as both a local package manager and a protocol to share packages across the network.
/// packages are apps; apps are packages. we use an onchain app listing contract to determine
@ -73,25 +69,20 @@ struct PackageListing {
// app store API
//
/// Remote requests, those sent between instantiations of this process
/// on different nodes, take this form. Will add more to enum in the future
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)] // untagged as a meta-type for all requests
pub enum Req {
LocalRequest(LocalRequest),
RemoteRequest(RemoteRequest),
FTWorkerCommand(FTWorkerCommand),
FTWorkerResult(FTWorkerResult),
pub enum RemoteRequest {
/// no payload; request a package from a node
/// remote node must return RemoteResponse::DownloadApproved,
/// at which point requester can expect a FTWorkerRequest::Receive
Download(PackageId),
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)] // untagged as a meta-type for all responses
pub enum Resp {
RemoteResponse(RemoteResponse),
FTWorkerResult(FTWorkerResult),
// note that we do not need to ourselves handle local responses, as
// those are given to others rather than received.
NewPackageResponse(NewPackageResponse),
DownloadResponse(DownloadResponse),
InstallResponse(InstallResponse),
pub enum RemoteResponse {
DownloadApproved,
DownloadDenied, // TODO expand on why
}
/// Local requests take this form.
@ -112,28 +103,23 @@ pub enum LocalRequest {
/// no payload; select a downloaded package and install it
/// if requested, will return an InstallResponse indicating success/failure
Install(PackageId),
/// no payload; select an installed package and uninstall it
/// no response will be given
/// Takes no payload; Select an installed package and uninstall it.
/// This will kill the processes in the **manifest** of the package,
/// but not the processes that were spawned by those processes! Take
/// care to kill those processes yourself. This will also delete the drive
/// containing the source code for this package. This does not guarantee
/// that other data created by this package will be removed from places such
/// as the key-value store.
Uninstall(PackageId),
/// no payload; select a downloaded package and delete it
/// no response will be given
Delete(PackageId),
}
/// Remote requests, those sent between instantiations of this process
/// on different nodes, take this form.
/// Local responses take this form.
#[derive(Debug, Serialize, Deserialize)]
pub enum RemoteRequest {
/// no payload; request a package from a node
/// remote node must return RemoteResponse::DownloadApproved,
/// at which point requester can expect a FTWorkerRequest::Receive
Download(PackageId),
}
#[derive(Debug, Serialize, Deserialize)]
pub enum RemoteResponse {
DownloadApproved,
DownloadDenied, // TODO expand on why
pub enum LocalResponse {
NewPackageResponse(NewPackageResponse),
DownloadResponse(DownloadResponse),
InstallResponse(InstallResponse),
UninstallResponse(UninstallResponse),
}
// TODO for all: expand these to elucidate why something failed
@ -157,39 +143,59 @@ pub enum InstallResponse {
Failure,
}
impl Guest for Component {
fn init(our: String) {
let our = Address::from_str(&our).unwrap();
// begin by granting messaging capabilities to http_server and terminal,
// so that they can send us requests.
grant_messaging(
&our,
vec![
ProcessId::new(Some("http_server"), "sys", "uqbar"),
ProcessId::new(Some("terminal"), "terminal", "uqbar"),
ProcessId::new(Some("vfs"), "sys", "uqbar"),
],
);
println!("{}: start", our.process);
#[derive(Debug, Serialize, Deserialize)]
pub enum UninstallResponse {
Success,
Failure,
}
// internal types
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)] // untagged as a meta-type for all incoming requests
pub enum Req {
LocalRequest(LocalRequest),
RemoteRequest(RemoteRequest),
FTWorkerCommand(FTWorkerCommand),
FTWorkerResult(FTWorkerResult),
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)] // untagged as a meta-type for all incoming responses
pub enum Resp {
RemoteResponse(RemoteResponse),
FTWorkerResult(FTWorkerResult),
}
#[derive(Debug, Serialize, Deserialize)]
struct ManifestCap {
process: String,
params: Value,
}
// /m our@main:app_store:ben.uq {"Download": {"package": {"package_name": "sdapi", "publisher_node": "benjammin.uq"}, "install_from": "testnode107.uq"}}
// /m our@main:app_store:ben.uq {"Install": {"package_name": "sdapi", "publisher_node": "benjammin.uq"}}
call_init!(init);
fn init(our: Address) {
println!("{}: running", our.process);
// load in our saved state or initalize a new one if none exists
let mut state =
get_typed_state(|bytes| Ok(bincode::deserialize(bytes)?)).unwrap_or(State {
let mut state = get_typed_state(|bytes| Ok(bincode::deserialize(bytes)?)).unwrap_or(State {
packages: HashMap::new(),
requested_packages: HashSet::new(),
});
// active the main messaging loop: handle requests and responses
loop {
match await_message() {
Err(send_error) => {
println!("{our}: got network error: {send_error:?}");
continue;
println!("app store: got network error: {send_error:?}");
}
Ok(message) => {
if let Err(e) = handle_message(&our, &mut state, &message) {
println!("app store: error handling message: {:?}", e)
}
Ok(message) => match handle_message(&our, &mut state, &message) {
Ok(()) => {}
Err(e) => println!("app-store: error handling message: {:?}", e),
},
}
}
}
@ -203,36 +209,25 @@ fn handle_message(our: &Address, mut state: &mut State, message: &Message) -> an
ipc,
..
} => {
match &serde_json::from_slice::<Req>(&ipc) {
Ok(Req::LocalRequest(local_request)) => {
match handle_local_request(&our, &source, local_request, &mut state) {
Ok(None) => return Ok(()),
Ok(Some(resp)) => {
match &serde_json::from_slice::<Req>(&ipc)? {
Req::LocalRequest(local_request) => {
if our.node != source.node {
return Err(anyhow::anyhow!("local request from non-local node"));
}
let resp = handle_local_request(&our, local_request, &mut state);
if expects_response.is_some() {
Response::new().ipc(serde_json::to_vec(&resp)?).send()?;
}
}
Err(err) => {
println!("app-store: local request error: {:?}", err);
}
}
}
Ok(Req::RemoteRequest(remote_request)) => {
match handle_remote_request(&our, &source, remote_request, &mut state) {
Ok(None) => return Ok(()),
Ok(Some(resp)) => {
Req::RemoteRequest(remote_request) => {
let resp = handle_remote_request(&our, &source, remote_request, &mut state);
if expects_response.is_some() {
Response::new().ipc(serde_json::to_vec(&resp)?).send()?;
}
}
Err(err) => {
println!("app-store: remote request error: {:?}", err);
}
}
}
Ok(Req::FTWorkerResult(FTWorkerResult::ReceiveSuccess(name))) => {
Req::FTWorkerResult(FTWorkerResult::ReceiveSuccess(name)) => {
// do with file what you'd like here
println!("file_transfer: successfully received {:?}", name);
println!("app store: successfully received {:?}", name);
// remove leading / and .zip from file name to get package ID
let package_id = match PackageId::from_str(name[1..].trim_end_matches(".zip")) {
Ok(package_id) => package_id,
@ -241,6 +236,7 @@ fn handle_message(our: &Address, mut state: &mut State, message: &Message) -> an
return Err(anyhow::anyhow!(e));
}
};
// only install the app if we actually requested it
if state.requested_packages.remove(&package_id) {
// auto-take zip from payload and request ourself with New
Request::new()
@ -249,43 +245,37 @@ fn handle_message(our: &Address, mut state: &mut State, message: &Message) -> an
.ipc(serde_json::to_vec(&Req::LocalRequest(
LocalRequest::NewPackage {
package: package_id,
mirror: true,
mirror: true, // can turn off auto-mirroring
},
))?)
.send()?;
crate::set_state(&bincode::serialize(state)?);
}
}
Ok(Req::FTWorkerCommand(_)) => {
spawn_receive_transfer(&our, &ipc);
Req::FTWorkerResult(r) => {
println!("app store: got ft_worker result: {r:?}");
}
e => {
return Err(anyhow::anyhow!(
"app store bad request: {:?}, error {:?}",
ipc,
e
))
Req::FTWorkerCommand(_) => {
spawn_receive_transfer(&our, &ipc)?;
}
}
}
Message::Response { ipc, context, .. } => match &serde_json::from_slice::<Resp>(&ipc) {
Ok(Resp::RemoteResponse(remote_response)) => match remote_response {
Message::Response { ipc, context, .. } => match &serde_json::from_slice::<Resp>(&ipc)? {
Resp::RemoteResponse(remote_response) => match remote_response {
RemoteResponse::DownloadApproved => {
println!("app store: download approved, should be starting");
println!("app store: download approved");
}
RemoteResponse::DownloadDenied => {
println!("app store: could not download package from that node!");
}
},
Ok(Resp::FTWorkerResult(ft_worker_result)) => {
let Ok(context) =
serde_json::from_slice::<FileTransferContext>(&context.as_ref().unwrap())
else {
return Err(anyhow::anyhow!("file_transfer: got weird local request"));
};
Resp::FTWorkerResult(ft_worker_result) => {
let context =
serde_json::from_slice::<FileTransferContext>(&context.as_ref().unwrap())?;
match ft_worker_result {
FTWorkerResult::SendSuccess => {
println!(
"file_transfer: successfully shared app {} in {:.4}s",
"app store: successfully shared app {} in {:.4}s",
context.file_name,
std::time::SystemTime::now()
.duration_since(context.start_time)
@ -293,39 +283,76 @@ fn handle_message(our: &Address, mut state: &mut State, message: &Message) -> an
.as_secs_f64(),
);
}
e => return Err(anyhow::anyhow!("file_transfer: {:?}", e)),
e => return Err(anyhow::anyhow!("app store: ft_worker gave us {e:?}")),
}
}
_ => return Err(anyhow::anyhow!("bad response from file transfer worker")),
},
}
Ok(())
}
fn handle_local_request(
our: &Address,
source: &Address,
request: &LocalRequest,
state: &mut State,
) -> anyhow::Result<Option<Resp>> {
if our.node != source.node {
return Err(anyhow::anyhow!("local request from non-local node"));
}
/// only `our.node` can call this
fn handle_local_request(our: &Address, request: &LocalRequest, state: &mut State) -> LocalResponse {
match request {
LocalRequest::NewPackage { package, mirror } => {
match handle_new_package(our, package, *mirror, state) {
Ok(()) => LocalResponse::NewPackageResponse(NewPackageResponse::Success),
Err(_) => LocalResponse::NewPackageResponse(NewPackageResponse::Failure),
}
}
LocalRequest::Download {
package,
install_from,
} => LocalResponse::DownloadResponse(
match Request::new()
.target((install_from.as_str(), our.process.clone()))
.inherit(true)
.ipc(serde_json::to_vec(&RemoteRequest::Download(package.clone())).unwrap())
.send_and_await_response(5)
{
Ok(Ok(Message::Response { ipc, .. })) => {
match serde_json::from_slice::<Resp>(&ipc) {
Ok(Resp::RemoteResponse(RemoteResponse::DownloadApproved)) => {
state.requested_packages.insert(package.clone());
crate::set_state(&bincode::serialize(&state).unwrap());
DownloadResponse::Started
}
_ => DownloadResponse::Failure,
}
}
_ => DownloadResponse::Failure,
},
),
LocalRequest::Install(package) => match handle_install(our, package) {
Ok(()) => LocalResponse::InstallResponse(InstallResponse::Success),
Err(_) => LocalResponse::InstallResponse(InstallResponse::Failure),
},
LocalRequest::Uninstall(package) => match handle_uninstall(package) {
Ok(()) => LocalResponse::UninstallResponse(UninstallResponse::Success),
Err(_) => LocalResponse::UninstallResponse(UninstallResponse::Failure),
},
}
}
fn handle_new_package(
our: &Address,
package: &PackageId,
mirror: bool,
state: &mut State,
) -> anyhow::Result<()> {
let Some(mut payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
let drive = format!("/{}/pkg", package);
// create a new drive for this package in VFS
Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: drive.clone(),
action: kt::VfsAction::CreateDrive,
})?)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
// produce the version hash for this new package
let mut hasher = sha2::Sha256::new();
@ -335,44 +362,43 @@ fn handle_local_request(
// add zip bytes
payload.mime = Some("application/zip".to_string());
let response = Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: drive.clone(),
action: kt::VfsAction::AddZip,
})?)
.payload(payload.clone())
.send_and_await_response(5)?.unwrap();
let Message::Response { ipc: ref vfs_ipc, .. } = response else {
panic!("app_store: send_and_await_response must return Response");
};
let vfs_ipc = serde_json::from_slice::<serde_json::Value>(vfs_ipc)?;
.send_and_await_response(5)??;
let vfs_ipc = serde_json::from_slice::<serde_json::Value>(response.ipc())?;
if vfs_ipc == serde_json::json!({"Err": "NoCap"}) {
return Err(anyhow::anyhow!("cannot add NewPackage: do not have capability to access vfs"));
return Err(anyhow::anyhow!(
"cannot add NewPackage: do not have capability to access vfs"
));
}
// save the zip file itself in VFS for sharing with other nodes
// call it <package>.zip
let zip_path = format!("{}/{}.zip", drive.clone(), package);
Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.inherit(true)
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: zip_path,
action: kt::VfsAction::ReWrite,
})?)
.payload(payload)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
let metadata_path = format!("{}/metadata.json", drive.clone());
// now, read the pkg contents to create our own listing and state,
// such that we can mirror this package to others.
Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: metadata_path,
action: kt::VfsAction::Read,
})?)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no metadata found!"));
};
@ -391,49 +417,23 @@ fn handle_local_request(
let package_state = PackageState {
mirrored_from: our.node.clone(),
listing_data,
mirroring: *mirror,
mirroring: mirror,
auto_update: true,
};
state.packages.insert(package.clone(), package_state);
crate::set_state(&bincode::serialize(state)?);
Ok(Some(Resp::NewPackageResponse(NewPackageResponse::Success)))
}
LocalRequest::Download {
package,
install_from,
} => Ok(Some(Resp::DownloadResponse(
match Request::new()
.target(Address::new(install_from, our.process.clone()))
.inherit(true)
.ipc(serde_json::to_vec(&RemoteRequest::Download(
package.clone(),
))?)
.send_and_await_response(5)
{
Ok(Ok(Message::Response { ipc, .. })) => {
let resp = serde_json::from_slice::<Resp>(&ipc)?;
match resp {
Resp::RemoteResponse(RemoteResponse::DownloadApproved) => {
state.requested_packages.insert(package.clone());
crate::set_state(&bincode::serialize(&state)?);
DownloadResponse::Started
}
_ => DownloadResponse::Failure,
}
}
_ => DownloadResponse::Failure,
},
))),
LocalRequest::Install(package) => {
crate::set_state(&bincode::serialize(state).unwrap());
Ok(())
}
fn handle_install(our: &Address, package: &PackageId) -> anyhow::Result<()> {
let drive_path = format!("/{}/pkg", package);
Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: format!("{}/manifest.json", drive_path),
action: kt::VfsAction::Read,
})?)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
let Some(payload) = get_payload() else {
return Err(anyhow::anyhow!("no payload"));
};
@ -447,7 +447,7 @@ fn handle_local_request(
"drive": drive_path,
}))?,
) else {
return Err(anyhow::anyhow!("app-store: no read cap"));
return Err(anyhow::anyhow!("app store: no read cap"));
};
let Some(write_cap) = get_capability(
&Address::new(&our.node, ("vfs", "sys", "uqbar")),
@ -456,13 +456,13 @@ fn handle_local_request(
"drive": drive_path,
}))?,
) else {
return Err(anyhow::anyhow!("app-store: no write cap"));
return Err(anyhow::anyhow!("app store: no write cap"));
};
let Some(networking_cap) = get_capability(
&Address::new(&our.node, ("kernel", "sys", "uqbar")),
&"\"network\"".to_string(),
) else {
return Err(anyhow::anyhow!("app-store: no net cap"));
return Err(anyhow::anyhow!("app store: no net cap"));
};
// first, for each process in manifest, initialize it
// then, once all have been initialized, grant them requested caps
@ -477,33 +477,31 @@ fn handle_local_request(
// build initial caps
let mut initial_capabilities: HashSet<kt::SignedCapability> = HashSet::new();
if entry.request_networking {
initial_capabilities
.insert(kt::de_wit_signed_capability(networking_cap.clone()));
initial_capabilities.insert(kt::de_wit_signed_capability(networking_cap.clone()));
}
initial_capabilities.insert(kt::de_wit_signed_capability(read_cap.clone()));
initial_capabilities.insert(kt::de_wit_signed_capability(write_cap.clone()));
let process_id = format!("{}:{}", entry.process_name, package);
let Ok(parsed_new_process_id) = ProcessId::from_str(&process_id) else {
return Err(anyhow::anyhow!("app-store: invalid process id!"));
return Err(anyhow::anyhow!("app store: invalid process id!"));
};
// kill process if it already exists
Request::new()
.target(Address::from_str("our@kernel:sys:uqbar")?)
.target(("our", "kernel", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::KernelCommand::KillProcess(
parsed_new_process_id.clone(),
))?)
.send()?;
let _bytes_response = Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: wasm_path.clone(),
action: kt::VfsAction::Read,
})?)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
Request::new()
.target(Address::from_str("our@kernel:sys:uqbar")?)
.target(("our", "kernel", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::KernelCommand::InitializeProcess {
id: parsed_new_process_id,
wasm_bytes_handle: wasm_path,
@ -512,8 +510,7 @@ fn handle_local_request(
public: entry.public,
})?)
.inherit(true)
.send_and_await_response(5)?
.unwrap();
.send_and_await_response(5)??;
}
for entry in &manifest {
let process_id = ProcessId::new(
@ -524,8 +521,7 @@ fn handle_local_request(
if let Some(to_request) = &entry.request_messaging {
for value in to_request {
let mut capability = None;
match value {
serde_json::Value::String(process_name) => {
if let serde_json::Value::String(process_name) = value {
if let Ok(parsed_process_id) = ProcessId::from_str(process_name) {
capability = get_capability(
&Address {
@ -535,32 +531,28 @@ fn handle_local_request(
&"\"messaging\"".into(),
);
}
}
serde_json::Value::Object(map) => {
if let Some(process_name) = map.get("process") {
if let Ok(parsed_process_id) =
ProcessId::from_str(&process_name.to_string())
{
if let Some(params) = map.get("params") {
} else {
let Ok(parsed) = serde_json::from_value::<ManifestCap>(value.to_owned()) else {
continue
};
if let Ok(parsed_process_id) = ProcessId::from_str(&parsed.process) {
capability = get_capability(
&Address {
node: our.node.clone(),
process: parsed_process_id.clone(),
},
&params.to_string(),
&parsed.params.to_string(),
);
}
}
}
}
_ => {
continue;
}
}
if let Some(cap) = capability {
share_capability(&process_id, &cap);
} else {
println!("app-store: no cap: {}, for {} to request!", value.to_string(), process_id);
println!(
"app store: no cap {} for {} to request!",
value.to_string(),
process_id
);
}
}
}
@ -607,29 +599,60 @@ fn handle_local_request(
if let Some(cap) = capability {
share_capability(&to_process.unwrap(), &cap);
} else {
println!("app-store: no cap: {}, for {} to grant!", value.to_string(), process_id);
println!(
"app store: no cap {} for {} to grant!",
value.to_string(),
process_id
);
}
}
}
Request::new()
.target(Address::from_str("our@kernel:sys:uqbar")?)
.target(("our", "kernel", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::KernelCommand::RunProcess(
process_id,
))?)
.send_and_await_response(5)?
.unwrap();
}
Ok(Some(Resp::InstallResponse(InstallResponse::Success)))
}
LocalRequest::Uninstall(_package) => {
// TODO
Ok(None)
}
LocalRequest::Delete(_package) => {
// TODO
Ok(None)
.send_and_await_response(5)??;
}
Ok(())
}
fn handle_uninstall(package: &PackageId) -> anyhow::Result<()> {
let drive_path = format!("/{}/pkg", package);
Request::new()
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: format!("{}/manifest.json", drive_path),
action: kt::VfsAction::Read,
})?)
.send_and_await_response(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<kt::PackageManifestEntry>>(&manifest)?;
// reading from the package manifest, kill every process
for entry in &manifest {
let process_id = format!("{}:{}", entry.process_name, package);
let Ok(parsed_new_process_id) = ProcessId::from_str(&process_id) else {
continue
};
Request::new()
.target(("our", "kernel", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::KernelCommand::KillProcess(
parsed_new_process_id,
))?)
.send()?;
}
// then, delete the drive
Request::new()
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: drive_path,
action: kt::VfsAction::RemoveDirAll,
})?)
.send_and_await_response(5)??;
Ok(())
}
fn handle_remote_request(
@ -637,30 +660,32 @@ fn handle_remote_request(
source: &Address,
request: &RemoteRequest,
state: &mut State,
) -> anyhow::Result<Option<Resp>> {
) -> Resp {
match request {
RemoteRequest::Download(package) => {
let Some(package_state) = state.packages.get(&package) else {
return Ok(Some(Resp::RemoteResponse(RemoteResponse::DownloadDenied)));
return Resp::RemoteResponse(RemoteResponse::DownloadDenied);
};
if !package_state.mirroring {
return Ok(Some(Resp::RemoteResponse(RemoteResponse::DownloadDenied)));
return Resp::RemoteResponse(RemoteResponse::DownloadDenied);
}
// get the .zip from VFS and attach as payload to response
let drive_name = format!("/{}/pkg", package);
let file_path = format!("/{}.zip", drive_name);
Request::new()
.target(Address::from_str("our@vfs:sys:uqbar")?)
let file_path = format!("/{}/pkg/{}.zip", package, package);
let Ok(Ok(_)) = Request::new()
.target(("our", "vfs", "sys", "uqbar"))
.ipc(serde_json::to_vec(&kt::VfsRequest {
path: file_path,
action: kt::VfsAction::Read,
})?)
.send_and_await_response(5)?
.unwrap();
// transfer will inherit the payload bytes we receive from VFS
}).unwrap())
.send_and_await_response(5) else {
return Resp::RemoteResponse(RemoteResponse::DownloadDenied);
};
// transfer will *inherit* the payload bytes we receive from VFS
let file_name = format!("/{}.zip", package);
spawn_transfer(&our, &file_name, None, &source);
Ok(Some(Resp::RemoteResponse(RemoteResponse::DownloadApproved)))
match spawn_transfer(&our, &file_name, None, 60, &source) {
Ok(()) => Resp::RemoteResponse(RemoteResponse::DownloadApproved),
Err(_e) => Resp::RemoteResponse(RemoteResponse::DownloadDenied),
}
}
}
}

36
modules/app_store/ft_worker/Cargo.lock generated
View File

@ -58,7 +58,7 @@ dependencies = [
[[package]]
name = "ft_worker"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"anyhow",
"bincode",
@ -157,6 +157,22 @@ version = "0.4.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f"
[[package]]
name = "mime"
version = "0.3.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
[[package]]
name = "mime_guess"
version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4192263c238a5f0d0c6bfd21f336a313a4ce1c450542449ca191bb657b4642ef"
dependencies = [
"mime",
"unicase",
]
[[package]]
name = "percent-encoding"
version = "2.3.1"
@ -321,6 +337,15 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "unicase"
version = "2.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7d2d4dafb69621809a81864c9c1b864479e1235c0dd4e199924b9742439ed89"
dependencies = [
"version_check",
]
[[package]]
name = "unicode-bidi"
version = "0.3.14"
@ -357,11 +382,12 @@ checksum = "f962df74c8c05a667b5ee8bcf162993134c104e96440b663c8daa176dc772d8c"
[[package]]
name = "uqbar_process_lib"
version = "0.4.0"
source = "git+ssh://git@github.com/uqbar-dao/process_lib.git?rev=b09d987#b09d9875edce1a230549cf56cf088f95e38d4abd"
source = "git+ssh://git@github.com/uqbar-dao/process_lib.git?rev=2d17d75#2d17d75152e55ef3ed417c79312e209ca45b8dbb"
dependencies = [
"anyhow",
"bincode",
"http",
"mime_guess",
"rand",
"serde",
"serde_json",
@ -381,6 +407,12 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "version_check"
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"

4
modules/app_store/ft_worker/Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "ft_worker"
version = "0.1.0"
version = "0.2.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@ -16,7 +16,7 @@ bincode = "1.3.3"
rand = "0.8"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
uqbar_process_lib = { git = "ssh://git@github.com/uqbar-dao/process_lib.git", rev = "b09d987" }
uqbar_process_lib = { git = "ssh://git@github.com/uqbar-dao/process_lib.git", rev = "2d17d75" }
wit-bindgen = { git = "https://github.com/bytecodealliance/wit-bindgen", rev = "efcc759" }
[lib]

100
modules/app_store/ft_worker/src/ft_worker_lib.rs
View File

@ -1,5 +1,5 @@
use serde::{Deserialize, Serialize};
use uqbar_process_lib::uqbar::process::standard::*;
use uqbar_process_lib::*;
#[derive(Debug, Serialize, Deserialize)]
pub struct FileTransferContext {
@ -13,9 +13,9 @@ pub struct FileTransferContext {
/// in order to prompt them to spawn a worker
#[derive(Debug, Serialize, Deserialize)]
pub enum FTWorkerCommand {
/// make sure to attach file itself as payload
Send {
// make sure to attach file itself as payload
target: String, // annoying, but this is Address
target: Address,
file_name: String,
timeout: u64,
},
@ -32,10 +32,12 @@ pub enum FTWorkerCommand {
#[derive(Debug, Serialize, Deserialize)]
pub enum FTWorkerResult {
SendSuccess,
ReceiveSuccess(String), // name of file, bytes in payload
/// string is name of file. bytes in payload
ReceiveSuccess(String),
Err(TransferError),
}
/// the possible errors that can be returned to the parent inside `FTWorkerResult`
#[derive(Debug, Serialize, Deserialize)]
pub enum TransferError {
TargetOffline,
@ -44,47 +46,48 @@ pub enum TransferError {
SourceFailed,
}
/// A helper function to spawn a worker and initialize a file transfer.
/// The outcome will be sent as an [`FTWorkerResult`] to the caller process.
///
/// if `file_bytes` is None, expects to inherit payload!
#[allow(dead_code)]
pub fn spawn_transfer(
our: &Address,
file_name: &str,
file_bytes: Option<Vec<u8>>, // if None, expects to inherit payload!
file_bytes: Option<Vec<u8>>,
timeout: u64,
to_addr: &Address,
) {
) -> anyhow::Result<()> {
let transfer_id: u64 = rand::random();
// spawn a worker and tell it to send the file
let Ok(worker_process_id) = spawn(
Some(&transfer_id.to_string()),
"/ft_worker.wasm".into(),
&OnExit::None, // can set message-on-panic here
Some(transfer_id.to_string().as_str()),
&format!("{}/pkg/ft_worker.wasm", our.package_id()),
OnExit::None, // can set message-on-panic here
&Capabilities::All,
false, // not public
) else {
print_to_terminal(0, "file_transfer: failed to spawn worker!");
return;
return Err(anyhow::anyhow!("failed to spawn ft_worker!"));
};
// tell the worker what to do
let payload_or_inherit = match file_bytes {
Some(bytes) => Some(Payload { mime: None, bytes }),
None => None,
};
send_request(
&Address {
node: our.node.clone(),
process: worker_process_id,
},
&Request {
inherit: !payload_or_inherit.is_some(),
expects_response: Some(61),
ipc: serde_json::to_vec(&FTWorkerCommand::Send {
target: to_addr.to_string(),
let mut req = Request::new()
.target((our.node.as_ref(), worker_process_id))
.inherit(!payload_or_inherit.is_some())
.expects_response(timeout + 1) // don't call with 2^64 lol
.ipc(
serde_json::to_vec(&FTWorkerCommand::Send {
target: to_addr.clone(),
file_name: file_name.into(),
timeout: 60,
timeout,
})
.unwrap(),
metadata: None,
},
Some(
&serde_json::to_vec(&FileTransferContext {
)
.context(
serde_json::to_vec(&FileTransferContext {
file_name: file_name.into(),
file_size: match &payload_or_inherit {
Some(p) => Some(p.bytes.len() as u64),
@ -93,39 +96,36 @@ pub fn spawn_transfer(
start_time: std::time::SystemTime::now(),
})
.unwrap(),
),
payload_or_inherit.as_ref(),
);
if let Some(payload) = payload_or_inherit {
req = req.payload(payload);
}
req.send()
}
pub fn spawn_receive_transfer(our: &Address, ipc: &[u8]) {
/// A helper function to allow a process to easily handle an incoming transfer
/// from an ft_worker. Call this when you get the initial [`FTWorkerCommand::Receive`]
/// and let it do the rest. The outcome will be sent as an [`FTWorkerResult`] inside
/// a Response to the caller.
#[allow(dead_code)]
pub fn spawn_receive_transfer(our: &Address, ipc: &[u8]) -> anyhow::Result<()> {
let Ok(FTWorkerCommand::Receive { transfer_id, .. }) = serde_json::from_slice(ipc) else {
print_to_terminal(0, "file_transfer: got weird request");
return;
return Err(anyhow::anyhow!("spawn_receive_transfer: got malformed request"));
};
let Ok(worker_process_id) = spawn(
Some(&transfer_id.to_string()),
"/ft_worker.wasm".into(),
&OnExit::None, // can set message-on-panic here
Some(transfer_id.to_string().as_str()),
&format!("{}/pkg/ft_worker.wasm", our.package_id()),
OnExit::None, // can set message-on-panic here
&Capabilities::All,
false, // not public
) else {
print_to_terminal(0, "file_transfer: failed to spawn worker!");
return;
return Err(anyhow::anyhow!("failed to spawn ft_worker!"));
};
// forward receive command to worker
send_request(
&Address {
node: our.node.clone(),
process: worker_process_id,
},
&Request {
inherit: true,
expects_response: None,
ipc: ipc.to_vec(),
metadata: None,
},
None,
None,
);
Request::new()
.target((our.node.as_ref(), worker_process_id))
.inherit(true)
.ipc(ipc)
.send()
}

213
modules/app_store/ft_worker/src/lib.rs
View File

@ -1,8 +1,6 @@
use serde::{Deserialize, Serialize};
//use uqbar_process_lib::uqbar::process::standard::*;
use uqbar_process_lib::uqbar::process::standard::{Message as StdMessage, Request as StdRequest, Response as StdResponse, SendErrorKind};
use uqbar_process_lib::{await_message, get_payload, print_to_terminal, send_and_await_response, send_request, send_response, Address, Message, Payload};
use uqbar_process_lib::*;
use uqbar_process_lib::println;
mod ft_worker_lib;
use ft_worker_lib::*;
@ -22,12 +20,9 @@ pub enum FTWorkerProtocol {
Finished,
}
struct Component;
impl Guest for Component {
fn init(our: String) {
let our = Address::from_str(&our).unwrap();
print_to_terminal(1, &format!("{}: start", our.process));
call_init!(init);
fn init(our: Address) {
let Ok(Message::Request { source: parent_process, ipc, .. }) = await_message() else {
panic!("ft_worker: got bad init message");
};
@ -35,129 +30,106 @@ impl Guest for Component {
let command = serde_json::from_slice::<FTWorkerCommand>(&ipc)
.expect("ft_worker: got unparseable init message");
match command {
let Some(result) = (match command {
FTWorkerCommand::Send {
target,
file_name,
timeout,
} => {
let transfer_id: u64 = our.process.process().parse().unwrap();
} => Some(handle_send(&our, &target, &file_name, timeout)),
FTWorkerCommand::Receive {
file_name,
total_chunks,
timeout,
..
} => handle_receive(parent_process, &file_name, total_chunks, timeout),
}) else { return };
Response::new()
.ipc(serde_json::to_vec(&result).unwrap())
.send()
.unwrap();
// job is done
}
fn handle_send(our: &Address, target: &Address, file_name: &str, timeout: u64) -> FTWorkerResult {
let transfer_id: u64 = our.process().parse().unwrap();
let Some(payload) = get_payload() else {
print_to_terminal(0, "FTWorker wasn't given payload, exiting");
return
println!("ft_worker: wasn't given payload!");
return FTWorkerResult::Err(TransferError::SourceFailed)
};
let file_bytes = payload.bytes;
let mut file_size = file_bytes.len() as u64;
let mut offset: u64 = 0;
let mut chunk_size: u64 = 1048576; // 1MB
let chunk_size: u64 = 1048576; // 1MB, can be changed
let total_chunks = (file_size as f64 / chunk_size as f64).ceil() as u64;
// send a file to another worker
// start by telling target to expect a file,
// then upon reciving affirmative response,
// send contents in chunks and wait for
// acknowledgement.
match send_and_await_response(
&Address::from_str(&target).unwrap(),
&StdRequest {
inherit: false,
expects_response: Some(timeout),
ipc: serde_json::to_vec(&FTWorkerCommand::Receive {
let Ok(Ok(response)) = Request::to(target.clone())
.ipc(serde_json::to_vec(&FTWorkerCommand::Receive {
transfer_id,
file_name,
file_name: file_name.to_string(),
file_size,
total_chunks,
timeout,
})
.unwrap(),
metadata: None,
},
None,
) {
Err(send_error) => {
respond_to_parent(FTWorkerResult::Err(match send_error.kind {
SendErrorKind::Offline => TransferError::TargetOffline,
SendErrorKind::Timeout => TransferError::TargetTimeout,
}))
}
Ok((opp_worker, StdMessage::Response((response, _)))) => {
let Ok(FTWorkerProtocol::Ready) = serde_json::from_slice(&response.ipc) else {
respond_to_parent(FTWorkerResult::Err(TransferError::TargetRejected));
return;
}).unwrap())
.send_and_await_response(timeout) else {
return FTWorkerResult::Err(TransferError::TargetOffline)
};
let opp_worker = response.source();
let Ok(FTWorkerProtocol::Ready) = serde_json::from_slice(&response.ipc()) else {
return FTWorkerResult::Err(TransferError::TargetRejected)
};
// send file in chunks
loop {
if file_size < chunk_size {
// this is the last chunk, so we should expect a Finished response
chunk_size = file_size;
let payload = Payload {
let _ = Request::to(opp_worker.clone())
.ipc(vec![])
.payload(Payload {
mime: None,
bytes: file_bytes
[offset as usize..offset as usize + chunk_size as usize]
bytes: file_bytes[offset as usize..offset as usize + file_size as usize]
.to_vec(),
};
send_request(
&opp_worker,
&StdRequest {
inherit: false,
expects_response: Some(timeout),
ipc: vec![],
metadata: None,
},
None,
Some(&payload),
);
})
.expects_response(timeout)
.send();
break;
}
let payload = Payload {
let _ = Request::to(opp_worker.clone())
.ipc(vec![])
.payload(Payload {
mime: None,
bytes: file_bytes
[offset as usize..offset as usize + chunk_size as usize]
.to_vec(),
};
send_request(
&opp_worker,
&StdRequest {
inherit: false,
expects_response: None,
ipc: vec![],
metadata: None,
},
None,
Some(&payload),
);
bytes: file_bytes[offset as usize..offset as usize + chunk_size as usize].to_vec(),
})
.send();
file_size -= chunk_size;
offset += chunk_size;
}
// now wait for Finished response
let Ok(Message::Response { ipc, .. }) = await_message() else {
respond_to_parent(FTWorkerResult::Err(TransferError::TargetRejected));
return;
return FTWorkerResult::Err(TransferError::TargetRejected)
};
let Ok(FTWorkerProtocol::Finished) = serde_json::from_slice(&ipc) else {
respond_to_parent(FTWorkerResult::Err(TransferError::TargetRejected));
return;
return FTWorkerResult::Err(TransferError::TargetRejected)
};
// return success to parent
respond_to_parent(FTWorkerResult::SendSuccess);
}
_ => respond_to_parent(FTWorkerResult::Err(TransferError::TargetRejected)),
}
}
FTWorkerCommand::Receive {
file_name,
total_chunks,
timeout,
..
} => {
return FTWorkerResult::SendSuccess;
}
fn handle_receive(
parent_process: Address,
file_name: &str,
total_chunks: u64,
timeout: u64,
) -> Option<FTWorkerResult> {
// send Ready response to counterparty
send_response(
&StdResponse {
inherit: false,
ipc: serde_json::to_vec(&FTWorkerProtocol::Ready).unwrap(),
metadata: None,
},
None,
);
Response::new()
.ipc(serde_json::to_vec(&FTWorkerProtocol::Ready).unwrap())
.send()
.unwrap();
// receive a file from a worker, then send it to parent
// all messages will be chunks of file. when we receive the
// last chunk, send a Finished message to sender and Success to parent.
@ -166,16 +138,13 @@ impl Guest for Component {
let start_time = std::time::Instant::now();
loop {
let Ok(Message::Request { .. }) = await_message() else {
respond_to_parent(FTWorkerResult::Err(TransferError::SourceFailed));
return;
return Some(FTWorkerResult::Err(TransferError::SourceFailed))
};
if start_time.elapsed().as_secs() > timeout {
respond_to_parent(FTWorkerResult::Err(TransferError::SourceFailed));
return;
return Some(FTWorkerResult::Err(TransferError::SourceFailed))
}
let Some(payload) = get_payload() else {
respond_to_parent(FTWorkerResult::Err(TransferError::SourceFailed));
return;
return Some(FTWorkerResult::Err(TransferError::SourceFailed))
};
chunks_received += 1;
file_bytes.extend(payload.bytes);
@ -184,42 +153,18 @@ impl Guest for Component {
}
}
// send Finished message to sender
send_response(
&StdResponse {
inherit: false,
ipc: serde_json::to_vec(&FTWorkerProtocol::Finished).unwrap(),
metadata: None,
},
None,
);
Response::new()
.ipc(serde_json::to_vec(&FTWorkerProtocol::Finished).unwrap())
.send()
.unwrap();
// send Success message to parent
send_request(
&parent_process,
&StdRequest {
inherit: false,
expects_response: None,
ipc: serde_json::to_vec(&FTWorkerResult::ReceiveSuccess(file_name))
.unwrap(),
metadata: None,
},
None,
Some(&Payload {
Request::to(parent_process)
.ipc(serde_json::to_vec(&FTWorkerResult::ReceiveSuccess(file_name.to_string())).unwrap())
.payload(Payload {
mime: None,
bytes: file_bytes,
}),
);
}
}
}
}
fn respond_to_parent(result: FTWorkerResult) {
send_response(
&StdResponse {
inherit: false,
ipc: serde_json::to_vec(&result).unwrap(),
metadata: None,
},
None,
);
})
.send()
.unwrap();
None
}

6
modules/app_store/pkg/manifest.json
View File

@ -9,7 +9,6 @@
"filesystem:sys:uqbar",
"http_server:sys:uqbar",
"http_client:sys:uqbar",
"encryptor:sys:uqbar",
"net:sys:uqbar",
"vfs:sys:uqbar",
"kernel:sys:uqbar",
@ -21,6 +20,11 @@
}
}
],
"grant_messaging": [
"http_server:sys:uqbar",
"terminal:terminal:uqbar",
"vfs:sys:uqbar"
],
"public": false
}
]

4
modules/app_store/pkg/metadata.json
View File

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

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

@ -6,8 +6,7 @@
"request_networking": false,
"request_messaging": [
"http_bindings:http_bindings:uqbar",
"http_server:sys:uqbar",
"encryptor:sys:uqbar"
"http_server:sys:uqbar"
],
"public": false
}

4
src/kernel/process.rs
View File

@ -69,7 +69,7 @@ pub async fn send_and_await_response(
}
let id = process
.process
.handle_request(source, target, request, None, payload)
.send_request(source, target, request, None, payload)
.await;
match id {
Ok(id) => match process.process.get_specific_message_for_process(id).await {
@ -103,7 +103,7 @@ impl ProcessState {
/// will only fail if process does not have capability to send to target.
/// if the request has a timeout (expects response), start a task to track
/// that timeout and return timeout error if it expires.
pub async fn handle_request(
pub async fn send_request(
&mut self,
fake_source: Option<t::Address>, // only used when kernel steps in to get/set state
target: wit::Address,

4
src/kernel/standard_host.rs
View File

@ -579,7 +579,7 @@ impl StandardHost for process::ProcessWasi {
) -> Result<()> {
let id = self
.process
.handle_request(None, target, request, context, payload)
.send_request(None, target, request, context, payload)
.await;
match id {
Ok(_id) => Ok(()),
@ -599,7 +599,7 @@ impl StandardHost for process::ProcessWasi {
for request in requests {
let id = self
.process
.handle_request(None, request.0, request.1, request.2, request.3)
.send_request(None, request.0, request.1, request.2, request.3)
.await;
match id {
Ok(_id) => continue,

13
src/kv.rs
View File

@ -267,8 +267,11 @@ async fn handle_request(
}
}
KvAction::Backup => {
// loop through all db directories and backup.
//
// looping through open dbs and flushing their memtables
for db_ref in open_kvs.iter() {
let db = db_ref.value();
db.flush()?;
}
(serde_json::to_vec(&KvResponse::Ok).unwrap(), None)
}
};
@ -428,11 +431,7 @@ async fn check_caps(
Ok(())
}
KvAction::Backup { .. } => {
if source.process != *STATE_PROCESS_ID {
return Err(KvError::NoCap {
error: request.action.to_string(),
});
}
// caps
Ok(())
}
}

21
src/sqlite.rs
View File

@ -284,8 +284,17 @@ async fn handle_request(
(serde_json::to_vec(&SqliteResponse::Ok).unwrap(), None)
}
SqliteAction::Backup => {
// execute WAL flush.
//
for db_ref in open_dbs.iter() {
let db = db_ref.value().lock().await;
let result: rusqlite::Result<()> = db
.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(()))
.map(|_| ());
if let Err(e) = result {
return Err(SqliteError::RusqliteError {
error: e.to_string(),
});
}
}
(serde_json::to_vec(&SqliteResponse::Ok).unwrap(), None)
}
};
@ -448,11 +457,7 @@ async fn check_caps(
Ok(())
}
SqliteAction::Backup => {
if source.process != *STATE_PROCESS_ID {
return Err(SqliteError::NoCap {
error: request.action.to_string(),
});
}
// flushing WALs for backup
Ok(())
}
}
@ -531,7 +536,7 @@ fn make_error_message(our_name: String, km: &KernelMessage, error: SqliteError)
id: km.id,
source: Address {
node: our_name.clone(),
process: KV_PROCESS_ID.clone(),
process: SQLITE_PROCESS_ID.clone(),
},
target: match &km.rsvp {
None => km.source.clone(),

190
src/state.rs
View File

@ -218,15 +218,23 @@ async fn handle_request(
}
}
StateAction::Backup => {
// handle Backup action
println!("got backup");
let checkpoint_dir = format!("{}/kernel/checkpoint", &home_directory_path);
let checkpoint_dir = format!("{}/kernel/backup", &home_directory_path);
if Path::new(&checkpoint_dir).exists() {
let _ = fs::remove_dir_all(&checkpoint_dir).await;
fs::remove_dir_all(&checkpoint_dir).await?;
}
let checkpoint = Checkpoint::new(&db).unwrap();
checkpoint.create_checkpoint(&checkpoint_dir).unwrap();
let checkpoint = Checkpoint::new(&db).map_err(|e| StateError::RocksDBError {
action: "BackupCheckpointNew".into(),
error: e.to_string(),
})?;
checkpoint.create_checkpoint(&checkpoint_dir).map_err(|e| {
StateError::RocksDBError {
action: "BackupCheckpointCreate".into(),
error: e.to_string(),
}
})?;
(serde_json::to_vec(&StateResponse::Backup).unwrap(), None)
}
};
@ -348,7 +356,7 @@ async fn bootstrap(
let packages = get_zipped_packages().await;
for (package_name, mut package) in packages {
for (package_name, mut package) in packages.clone() {
// special case tester: only load it in if in simulation mode
if package_name == "tester" {
#[cfg(not(feature = "simulation-mode"))]
@ -488,58 +496,6 @@ async fn bootstrap(
}
}
// grant capabilities to other initially spawned processes, distro
if let Some(to_grant) = &entry.grant_messaging {
for value in to_grant {
let mut capability = None;
let mut to_process = None;
match value {
serde_json::Value::String(process_name) => {
if let Ok(parsed_process_id) = ProcessId::from_str(process_name) {
capability = Some(Capability {
issuer: Address {
node: our_name.to_string(),
process: ProcessId::from_str(process_name).unwrap(),
},
params: "\"messaging\"".into(),
});
to_process = Some(parsed_process_id);
}
}
serde_json::Value::Object(map) => {
if let Some(process_name) = map.get("process") {
if let Ok(parsed_process_id) =
ProcessId::from_str(&process_name.as_str().unwrap())
{
if let Some(params) = map.get("params") {
capability = Some(Capability {
issuer: Address {
node: our_name.to_string(),
process: ProcessId::from_str(
process_name.as_str().unwrap(),
)
.unwrap(),
},
params: params.to_string(),
});
to_process = Some(parsed_process_id);
}
}
}
}
_ => {
continue;
}
}
if let Some(cap) = capability {
if let Some(process) = process_map.get_mut(&to_process.unwrap()) {
process.capabilities.insert(cap);
}
}
}
}
if entry.request_networking {
requested_caps.insert(Capability {
issuer: Address {
@ -589,6 +545,114 @@ async fn bootstrap(
);
}
}
// second loop: go and grant_capabilities to processes
// can't do this in first loop because we need to have all processes in the map first
for (package_name, mut package) in packages {
// special case tester: only load it in if in simulation mode
if package_name == "tester" {
#[cfg(not(feature = "simulation-mode"))]
continue;
#[cfg(feature = "simulation-mode")]
{}
}
// get and read manifest.json
let Ok(mut package_manifest_zip) = package.by_name("manifest.json") else {
println!(
"fs: missing manifest for package {}, skipping",
package_name
);
continue;
};
let mut manifest_content = Vec::new();
package_manifest_zip
.read_to_end(&mut manifest_content)
.unwrap();
drop(package_manifest_zip);
let package_manifest = String::from_utf8(manifest_content)?;
let package_manifest = serde_json::from_str::<Vec<PackageManifestEntry>>(&package_manifest)
.expect("fs: manifest parse error");
// get and read metadata.json
let Ok(mut package_metadata_zip) = package.by_name("metadata.json") else {
println!(
"fs: missing metadata for package {}, skipping",
package_name
);
continue;
};
let mut metadata_content = Vec::new();
package_metadata_zip
.read_to_end(&mut metadata_content)
.unwrap();
drop(package_metadata_zip);
let package_metadata: serde_json::Value =
serde_json::from_slice(&metadata_content).expect("fs: metadata parse error");
println!("fs: found package metadata: {:?}\r", package_metadata);
let package_name = package_metadata["package"]
.as_str()
.expect("fs: metadata parse error: bad package name");
let package_publisher = package_metadata["publisher"]
.as_str()
.expect("fs: metadata parse error: bad publisher name");
// for each process-entry in manifest.json:
for entry in package_manifest {
let our_process_id = format!(
"{}:{}:{}",
entry.process_name, package_name, package_publisher
);
// grant capabilities to other initially spawned processes, distro
if let Some(to_grant) = &entry.grant_messaging {
for value in to_grant {
match value {
serde_json::Value::String(process_name) => {
if let Ok(parsed_process_id) = ProcessId::from_str(process_name) {
if let Some(process) = process_map.get_mut(&parsed_process_id) {
process.capabilities.insert(Capability {
issuer: Address {
node: our_name.to_string(),
process: ProcessId::from_str(&our_process_id).unwrap(),
},
params: "\"messaging\"".into(),
});
}
}
}
serde_json::Value::Object(map) => {
if let Some(process_name) = map.get("process") {
if let Ok(parsed_process_id) =
ProcessId::from_str(&process_name.as_str().unwrap())
{
if let Some(params) = map.get("params") {
if let Some(process) =
process_map.get_mut(&parsed_process_id)
{
process.capabilities.insert(Capability {
issuer: Address {
node: our_name.to_string(),
process: ProcessId::from_str(&our_process_id)
.unwrap(),
},
params: params.to_string(),
});
}
}
}
}
}
_ => {
continue;
}
}
}
}
}
}
Ok(())
}
@ -609,6 +673,14 @@ async fn get_zipped_packages() -> Vec<(String, zip::ZipArchive<std::io::Cursor<&
packages
}
impl From<std::io::Error> for StateError {
fn from(err: std::io::Error) -> Self {
StateError::IOError {
error: err.to_string(),
}
}
}
fn make_error_message(our_name: String, km: &KernelMessage, error: StateError) -> KernelMessage {
KernelMessage {
id: km.id,

1
src/types.rs
View File

@ -5,7 +5,6 @@ use std::collections::{HashMap, HashSet};
use thiserror::Error;
lazy_static::lazy_static! {
pub static ref ENCRYPTOR_PROCESS_ID: ProcessId = ProcessId::new(Some("encryptor"), "sys", "uqbar");
pub static ref ETH_RPC_PROCESS_ID: ProcessId = ProcessId::new(Some("eth_rpc"), "sys", "uqbar");
pub static ref HTTP_CLIENT_PROCESS_ID: ProcessId = ProcessId::new(Some("http_client"), "sys", "uqbar");
pub static ref HTTP_SERVER_PROCESS_ID: ProcessId = ProcessId::new(Some("http_server"), "sys", "uqbar");

63
src/vfs.rs
View File

@ -558,6 +558,25 @@ async fn check_caps(
let src_package_id = PackageId::new(source.process.package(), source.process.publisher());
let (send_cap_bool, recv_cap_bool) = tokio::sync::oneshot::channel();
// check for root cap (todo make temp buffer so this is more efficient?)
send_to_caps_oracle
.send(CapMessage::Has {
on: source.process.clone(),
cap: Capability {
issuer: Address {
node: our_node.clone(),
process: VFS_PROCESS_ID.clone(),
},
params: serde_json::to_string(&serde_json::json!({
"root": true,
}))
.unwrap(),
},
responder: send_cap_bool,
})
.await?;
let has_root_cap = recv_cap_bool.await?;
match &request.action {
VfsAction::CreateDir
| VfsAction::CreateDirAll
@ -579,25 +598,7 @@ async fn check_caps(
if src_package_id == package_id {
return Ok(());
}
send_to_caps_oracle
.send(CapMessage::Has {
on: source.process.clone(),
cap: Capability {
issuer: Address {
node: our_node.clone(),
process: VFS_PROCESS_ID.clone(),
},
params: serde_json::to_string(&serde_json::json!({
"kind": "write",
"drive": drive,
}))
.unwrap(),
},
responder: send_cap_bool,
})
.await?;
let has_cap = recv_cap_bool.await?;
if !has_cap {
if !has_root_cap {
return Err(VfsError::NoCap {
action: request.action.to_string(),
path: path.display().to_string(),
@ -617,6 +618,10 @@ async fn check_caps(
if src_package_id == package_id {
return Ok(());
}
if has_root_cap {
return Ok(());
}
let (send_cap_bool, recv_cap_bool) = tokio::sync::oneshot::channel();
send_to_caps_oracle
.send(CapMessage::Has {
on: source.process.clone(),
@ -645,25 +650,7 @@ async fn check_caps(
}
VfsAction::CreateDrive => {
if src_package_id != package_id {
// might have root caps
send_to_caps_oracle
.send(CapMessage::Has {
on: source.process.clone(),
cap: Capability {
issuer: Address {
node: our_node.clone(),
process: VFS_PROCESS_ID.clone(),
},
params: serde_json::to_string(&serde_json::json!({
"root": true,
}))
.unwrap(),
},
responder: send_cap_bool,
})
.await?;
let has_cap = recv_cap_bool.await?;
if !has_cap {
if !has_root_cap {
return Err(VfsError::NoCap {
action: request.action.to_string(),
path: path.display().to_string(),