uqbar-dao/nectar
1
1
Fork 0
mirror of https://github.com/uqbar-dao/nectar.git synced 2024-12-29 19:41:39 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #645 from kinode-dao/dr/fix-dumb-error-types

Browse Source 
error types final
This commit is contained in:
doria 2024-12-23 02:31:49 -05:00 committed by GitHub
commit 56fd25f941
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 109 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
kinode/src/http/client.rs
View File

@ -64,9 +64,7 @@ pub async fn http_client(
id,
rsvp.unwrap_or(source),
expects_response,
HttpClientError::BadRequest {
req: String::from_utf8(body).unwrap_or_default(),
},
HttpClientError::MalformedRequest,
send_to_loop.clone(),
)
.await;
@ -197,8 +195,8 @@ async fn connect_websocket(
};
let Ok(mut req) = url.clone().into_client_request() else {
return Err(HttpClientError::BadRequest {
req: "failed to parse url into client request".into(),
return Err(HttpClientError::WsOpenFailed {
url: url.to_string(),
});
};
@ -443,18 +441,16 @@ async fn handle_http_request(
}
// Add the headers
let Ok(request) = request_builder
let build = request_builder
.headers(deserialize_headers(req.headers))
.build()
else {
.build();
if let Err(e) = build {
http_error_message(
our,
id,
target,
expects_response,
HttpClientError::RequestFailed {
error: "failed to build request".into(),
},
HttpClientError::BuildRequestFailed(e.to_string()),
send_to_loop,
)
.await;
@ -462,7 +458,7 @@ async fn handle_http_request(
};
// Send the HTTP request
match client.execute(request).await {
match client.execute(build.unwrap()).await {
Ok(response) => {
// Handle the response and forward to the target process
let Ok(body) = serde_json::to_vec::<Result<HttpClientResponse, HttpClientError>>(&Ok(
@ -512,9 +508,7 @@ async fn handle_http_request(
id,
target,
expects_response,
HttpClientError::RequestFailed {
error: e.to_string(),
},
HttpClientError::ExecuteRequestFailed(e.to_string()),
send_to_loop,
)
.await;
@ -612,32 +606,24 @@ async fn send_ws_push(
ws_streams: WebSocketStreams,
) -> Result<HttpClientResponse, HttpClientError> {
let Some(mut ws_stream) = ws_streams.get_mut(&(target.process.clone(), channel_id)) else {
return Err(HttpClientError::WsPushFailed {
req: format!("channel_id {} not found", channel_id),
});
return Err(HttpClientError::WsPushUnknownChannel { channel_id });
};
let _ = match message_type {
WsMessageType::Text => {
let Some(blob) = blob else {
return Err(HttpClientError::WsPushFailed {
req: "no blob".into(),
});
return Err(HttpClientError::WsPushNoBlob);
};
let Ok(text) = String::from_utf8(blob.bytes) else {
return Err(HttpClientError::WsPushFailed {
req: "failed to convert blob to string".into(),
});
return Err(HttpClientError::WsPushBadText);
};
ws_stream.send(TungsteniteMessage::Text(text)).await
}
WsMessageType::Binary => {
let Some(blob) = blob else {
return Err(HttpClientError::WsPushFailed {
req: "no blob".into(),
});
return Err(HttpClientError::WsPushNoBlob);
};
ws_stream.send(TungsteniteMessage::Binary(blob.bytes)).await

62
kinode/src/http/server.rs
View File

@ -104,9 +104,7 @@ async fn send_push(
id,
source,
send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "Ping and Pong messages must be 125 bytes or less".to_string(),
}),
Err(HttpServerError::WsPingPongTooLong),
)
.await;
return true;
@ -130,9 +128,7 @@ async fn send_push(
id,
source,
send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "WebSocket channel not owned by this process".to_string(),
}),
Err(HttpServerError::WsChannelNotFound),
)
.await;
return true;
@ -144,9 +140,7 @@ async fn send_push(
id,
source.clone(),
send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "WebSocket channel closed".to_string(),
}),
Err(HttpServerError::WsChannelNotFound),
)
.await;
return true;
@ -157,9 +151,7 @@ async fn send_push(
id,
source.clone(),
send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "WebSocket channel not found".to_string(),
}),
Err(HttpServerError::WsChannelNotFound),
)
.await;
return true;
@ -1180,9 +1172,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::BadRequest {
req: String::from_utf8_lossy(body).to_string(),
}),
Err(HttpServerError::MalformedRequest),
)
.await;
return;
@ -1200,12 +1190,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::PathBindError {
error: format!(
"invalid source process (not Kimap safe): {}",
source
),
}),
Err(HttpServerError::InvalidSourceProcess),
)
.await;
return;
@ -1271,12 +1256,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::PathBindError {
error: format!(
"invalid source process (not Kimap safe): {}",
source
),
}),
Err(HttpServerError::InvalidSourceProcess),
)
.await;
return;
@ -1356,12 +1336,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::PathBindError {
error: format!(
"invalid source process (not Kimap safe): {}",
source
),
}),
Err(HttpServerError::InvalidSourceProcess),
)
.await;
return;
@ -1385,12 +1360,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::PathBindError {
error: format!(
"invalid source process (not Kimap safe): {}",
source
),
}),
Err(HttpServerError::InvalidSourceProcess),
)
.await;
return;
@ -1427,9 +1397,7 @@ async fn handle_app_message(
km.id,
km.source.clone(),
&send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "WebSocketOpen is not a valid request".to_string(),
}),
Err(HttpServerError::MalformedRequest),
)
.await;
}
@ -1475,10 +1443,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "Use WebSocketExtPushOutgoing, not WebSocketExtPushData"
.to_string(),
}),
Err(HttpServerError::MalformedRequest),
)
.await;
return;
@ -1490,10 +1455,7 @@ async fn handle_app_message(
km.id,
km.source,
&send_to_loop,
Err(HttpServerError::WebSocketPushError {
error: "WebSocket channel not owned by this process"
.to_string(),
}),
Err(HttpServerError::WsChannelNotFound),
)
.await;
return;

140
kinode/src/vfs.rs
View File

@ -287,14 +287,12 @@ async fn handle_request(
..
}) = km.message
else {
return Err(VfsError::BadRequest {
error: "not a request".into(),
});
// we got a response -- safe to ignore
return Ok(());
};
let request: VfsRequest = serde_json::from_slice(&body).map_err(|e| VfsError::BadJson {
error: e.to_string(),
})?;
let request: VfsRequest =
serde_json::from_slice(&body).map_err(|_| VfsError::MalformedRequest)?;
// special case for root reading list of all drives.
if request.action == VfsAction::ReadDir && request.path == "/" {
@ -336,10 +334,7 @@ async fn handle_request(
.await;
return Ok(());
} else {
return Err(VfsError::NoCap {
action: request.action.to_string(),
path: request.path,
});
return Err(VfsError::NoReadCap);
}
}
@ -348,7 +343,6 @@ async fn handle_request(
// must have prepended `/` here or else it messes up caps downstream, e.g. in run-tests
let drive = format!("/{package_id}/{drive}");
let action = request.action;
let path = PathBuf::from(&request.path);
if km.source.process != *KERNEL_PROCESS_ID {
check_caps(
@ -356,7 +350,6 @@ async fn handle_request(
&km.source,
&send_to_caps_oracle,
&action,
&path,
&drive,
&package_id,
vfs_path,
@ -406,9 +399,7 @@ async fn handle_request(
VfsAction::WriteAll => {
// doesn't create a file, writes at exact cursor.
let Some(blob) = km.lazy_load_blob else {
return Err(VfsError::BadRequest {
error: "blob needs to exist for WriteAll".into(),
});
return Err(VfsError::NoBlob);
};
let file = files.open_file(&path, false, false).await?;
let mut file = file.lock().await;
@ -417,18 +408,14 @@ async fn handle_request(
}
VfsAction::Write => {
let Some(blob) = km.lazy_load_blob else {
return Err(VfsError::BadRequest {
error: "blob needs to exist for Write".into(),
});
return Err(VfsError::NoBlob);
};
fs::write(&path, &blob.bytes).await?;
(VfsResponse::Ok, None)
}
VfsAction::Append => {
let Some(blob) = km.lazy_load_blob else {
return Err(VfsError::BadRequest {
error: "blob needs to exist for Append".into(),
});
return Err(VfsError::NoBlob);
};
let file = files.open_file(&path, false, false).await?;
let mut file = file.lock().await;
@ -526,29 +513,16 @@ async fn handle_request(
}
VfsAction::Rename { new_path } => {
let new_path = join_paths_safely(vfs_path, &new_path);
fs::rename(&path, new_path)
.await
.map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path,
})?;
fs::rename(&path, new_path).await?;
(VfsResponse::Ok, None)
}
VfsAction::CopyFile { new_path } => {
let new_path = join_paths_safely(vfs_path, &new_path);
fs::copy(&path, new_path)
.await
.map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path,
})?;
fs::copy(&path, new_path).await?;
(VfsResponse::Ok, None)
}
VfsAction::Metadata => {
let metadata = fs::metadata(&path).await.map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path,
})?;
let metadata = fs::metadata(&path).await?;
let file_type = get_file_type(&metadata);
let meta = FileMetadata {
len: metadata.len(),
@ -559,23 +533,13 @@ async fn handle_request(
VfsAction::Len => {
let file = files.open_file(&path, false, false).await?;
let file = file.lock().await;
let len = file
.metadata()
.await
.map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path,
})?
.len();
let len = file.metadata().await?.len();
(VfsResponse::Len(len), None)
}
VfsAction::SetLen(len) => {
let file = files.open_file(&path, false, false).await?;
let file = file.lock().await;
file.set_len(len).await.map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path,
})?;
file.set_len(len).await?;
(VfsResponse::Ok, None)
}
VfsAction::Hash => {
@ -597,9 +561,7 @@ async fn handle_request(
}
VfsAction::AddZip => {
let Some(blob) = km.lazy_load_blob else {
return Err(VfsError::BadRequest {
error: "blob needs to exist for AddZip".into(),
});
return Err(VfsError::NoBlob);
};
let file = std::io::Cursor::new(&blob.bytes);
let mut zip = match zip::ZipArchive::new(file) {
@ -620,10 +582,7 @@ async fn handle_request(
// Before any `.await`s are called since ZipFile is not
// Send and so does not play nicely with await
let (is_file, is_dir, local_path, file_contents) = {
let mut file = zip.by_index(i).map_err(|e| VfsError::IOError {
error: e.to_string(),
path: request.path.clone(),
})?;
let mut file = zip.by_index(i).map_err(|_| VfsError::UnzipError)?;
let is_file = file.is_file();
let is_dir = file.is_dir();
let mut file_contents = Vec::new();
@ -638,10 +597,7 @@ async fn handle_request(
} else if is_dir {
fs::create_dir_all(&local_path).await?;
} else {
return Err(VfsError::CreateDirError {
path: path.display().to_string(),
error: "vfs: zip with non-file non-dir".into(),
});
return Err(VfsError::UnzipError);
};
}
(VfsResponse::Ok, None)
@ -684,17 +640,13 @@ fn parse_package_and_drive(
// sanitize path..
let normalized_path = normalize_path(&joined_path);
if !normalized_path.starts_with(vfs_path) {
return Err(VfsError::BadRequest {
error: format!("input path tries to escape parent vfs directory: {path}"),
})?;
return Err(VfsError::MalformedRequest);
}
// extract original path.
let path = normalized_path
.strip_prefix(vfs_path)
.map_err(|_| VfsError::BadRequest {
error: format!("input path tries to escape parent vfs directory: {path}"),
})?
.map_err(|_| VfsError::MalformedRequest)?
.display()
.to_string();
@ -707,19 +659,13 @@ fn parse_package_and_drive(
parts.remove(0);
}
if parts.len() < 2 {
return Err(VfsError::ParseError {
error: "malformed path".into(),
path,
});
return Err(VfsError::MalformedRequest);
}
let package_id = match parts[0].parse::<PackageId>() {
Ok(id) => id,
Err(e) => {
return Err(VfsError::ParseError {
error: e.to_string(),
path,
})
Err(_) => {
return Err(VfsError::MalformedRequest);
}
};
@ -774,7 +720,6 @@ async fn check_caps(
source: &Address,
send_to_caps_oracle: &CapMessageSender,
action: &VfsAction,
path: &PathBuf,
drive: &str,
package_id: &PackageId,
vfs_path: &PathBuf,
@ -809,10 +754,7 @@ async fn check_caps(
if read_capability("", "", true, our_node, source, send_to_caps_oracle).await {
return Ok(());
}
return Err(VfsError::NoCap {
action: action.to_string(),
path: path.display().to_string(),
});
return Err(VfsError::NoWriteCap);
}
Ok(())
}
@ -835,10 +777,7 @@ async fn check_caps(
if read_capability("", "", true, our_node, source, send_to_caps_oracle).await {
return Ok(());
}
return Err(VfsError::NoCap {
action: action.to_string(),
path: path.display().to_string(),
});
return Err(VfsError::NoReadCap);
}
Ok(())
}
@ -867,10 +806,7 @@ async fn check_caps(
if read_capability("", "", true, our_node, source, send_to_caps_oracle).await {
return Ok(());
}
return Err(VfsError::NoCap {
action: action.to_string(),
path: path.display().to_string(),
});
return Err(VfsError::NoWriteCap);
}
// if they're within the same drive, no need for 2 caps checks
@ -892,10 +828,7 @@ async fn check_caps(
if read_capability("", "", true, our_node, source, send_to_caps_oracle).await {
return Ok(());
}
return Err(VfsError::NoCap {
action: action.to_string(),
path: path.display().to_string(),
});
return Err(VfsError::NoWriteCap);
}
Ok(())
}
@ -903,10 +836,7 @@ async fn check_caps(
if &src_package_id != package_id {
// check for root cap
if !read_capability("", "", true, our_node, source, send_to_caps_oracle).await {
return Err(VfsError::NoCap {
action: action.to_string(),
path: path.display().to_string(),
});
return Err(VfsError::NoWriteCap);
}
}
add_capability("read", &drive, &our_node, &source, send_to_caps_oracle).await?;
@ -958,20 +888,20 @@ async fn add_capability(
format!("{{\"kind\": \"{kind}\", \"drive\": \"{drive}\"}}"),
);
let (send_cap_bool, recv_cap_bool) = tokio::sync::oneshot::channel();
send_to_caps_oracle
let Ok(()) = send_to_caps_oracle
.send(CapMessage::Add {
on: source.process.clone(),
caps: vec![cap],
responder: Some(send_cap_bool),
})
.await?;
match recv_cap_bool.await? {
true => Ok(()),
false => Err(VfsError::NoCap {
action: "add_capability".to_string(),
path: drive.to_string(),
}),
}
.await
else {
return Err(VfsError::AddCapFailed);
};
let Ok(true) = recv_cap_bool.await else {
return Err(VfsError::AddCapFailed);
};
Ok(())
}
fn get_file_type(metadata: &std::fs::Metadata) -> FileType {

37
lib/src/http/client_types.rs
View File

@ -23,12 +23,11 @@ pub enum HttpClientAction {
},
}
/// HTTP Request type that can be shared over Wasm boundary to apps.
/// This is the one you send to the `http-client:distro:sys` service.
/// HTTP Request type contained in [`HttpClientAction::Http`].
///
/// BODY is stored in the lazy_load_blob, as bytes
///
/// TIMEOUT is stored in the message expect_response value
/// TIMEOUT is stored in the message's `expects_response` value
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OutgoingHttpRequest {
/// must parse to [`http::Method`]
@ -56,31 +55,37 @@ pub enum HttpClientRequest {
/// Response type received from the `http-client:distro:sys` service after
/// sending a successful [`HttpClientAction`] to it.
#[derive(Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum HttpClientResponse {
Http(HttpResponse),
WebSocketAck,
}
#[derive(Error, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, Error, Serialize, Deserialize)]
pub enum HttpClientError {
// HTTP errors
#[error("http-client: request is not valid HttpClientRequest: {req}.")]
BadRequest { req: String },
#[error("http-client: http method not supported: {method}.")]
#[error("request could not be deserialized to valid HttpClientRequest")]
MalformedRequest,
#[error("http method not supported: {method}")]
BadMethod { method: String },
#[error("http-client: url could not be parsed: {url}.")]
#[error("url could not be parsed: {url}")]
BadUrl { url: String },
#[error("http-client: http version not supported: {version}.")]
#[error("http version not supported: {version}")]
BadVersion { version: String },
#[error("http-client: failed to execute request {error}.")]
RequestFailed { error: String },
#[error("client failed to build request: {0}")]
BuildRequestFailed(String),
#[error("client failed to execute request: {0}")]
ExecuteRequestFailed(String),
// WebSocket errors
#[error("http-client: failed to open connection {url}.")]
#[error("could not open connection to {url}")]
WsOpenFailed { url: String },
#[error("http-client: failed to send message {req}.")]
WsPushFailed { req: String },
#[error("http-client: failed to close connection {channel_id}.")]
#[error("sent WebSocket push to unknown channel {channel_id}")]
WsPushUnknownChannel { channel_id: u32 },
#[error("WebSocket push failed because message had no blob attached")]
WsPushNoBlob,
#[error("WebSocket push failed because message type was Text, but blob was not a valid UTF-8 string")]
WsPushBadText,
#[error("failed to close connection {channel_id} because it was not open")]
WsCloseFailed { channel_id: u32 },
}

19
lib/src/http/server_types.rs
View File

@ -48,11 +48,12 @@ pub struct IncomingHttpRequest {
/// HTTP Response type that can be shared over Wasm boundary to apps.
/// Respond to [`IncomingHttpRequest`] with this type.
#[derive(Debug, Serialize, Deserialize)]
///
/// BODY is stored in the lazy_load_blob, as bytes
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct HttpResponse {
pub status: u16,
pub headers: HashMap<String, String>,
// BODY is stored in the lazy_load_blob, as bytes
}
#[derive(Debug, Serialize, Deserialize)]
@ -168,14 +169,16 @@ pub enum WsMessageType {
/// Part of the Response type issued by `http-server:distro:sys`
#[derive(Error, Debug, Serialize, Deserialize)]
pub enum HttpServerError {
#[error("request could not be parsed to HttpServerAction: {req}.")]
BadRequest { req: String },
#[error("request could not be deserialized to valid HttpServerRequest")]
MalformedRequest,
#[error("action expected blob")]
NoBlob,
#[error("path binding error: {error}")]
PathBindError { error: String },
#[error("WebSocket error: {error}")]
WebSocketPushError { error: String },
#[error("path binding error: invalid source process")]
InvalidSourceProcess,
#[error("WebSocket error: ping/pong message too long")]
WsPingPongTooLong,
#[error("WebSocket error: channel not found")]
WsChannelNotFound,
}
/// Structure sent from client websocket to this server upon opening a new connection.

2
lib/src/state.rs
View File

@ -2,6 +2,8 @@ use crate::types::core::ProcessId;
use serde::{Deserialize, Serialize};
use thiserror::Error;
// state:distro:sys is INTERNAL -- this interface is NEVER EXPOSED TO USERSPACE
/// IPC Requests for the state:distro:sys runtime module.
#[derive(Serialize, Deserialize, Debug)]
pub enum StateAction {

77
lib/src/vfs.rs
View File

@ -1,4 +1,3 @@
use crate::types::core::CapMessage;
use serde::{Deserialize, Serialize};
use thiserror::Error;
@ -81,72 +80,26 @@ pub enum VfsResponse {
#[derive(Error, Debug, Serialize, Deserialize)]
pub enum VfsError {
#[error("No capability for action {action} at path {path}")]
NoCap { action: String, path: String },
#[error("Bytes blob required for {action} at path {path}")]
BadBytes { action: String, path: String },
#[error("bad request error: {error}")]
BadRequest { error: String },
#[error("no write capability for requested drive")]
NoWriteCap,
#[error("no read capability for requested drive")]
NoReadCap,
#[error("failed to generate capability for new drive")]
AddCapFailed,
#[error("request could not be deserialized to valid VfsRequest")]
MalformedRequest,
#[error("request type used requires a blob")]
NoBlob,
#[error("error parsing path: {path}: {error}")]
ParseError { error: String, path: String },
#[error("IO error: {error}, at path {path}")]
IOError { error: String, path: String },
#[error("kernel capability channel error: {error}")]
CapChannelFail { error: String },
#[error("Bad JSON blob: {error}")]
BadJson { error: String },
#[error("File not found at path {path}")]
NotFound { path: String },
#[error("Creating directory failed at path: {path}: {error}")]
CreateDirError { path: String, error: String },
#[error("Other error: {error}")]
Other { error: String },
}
impl VfsError {
pub fn kind(&self) -> &str {
match *self {
VfsError::NoCap { .. } => "NoCap",
VfsError::BadBytes { .. } => "BadBytes",
VfsError::BadRequest { .. } => "BadRequest",
VfsError::ParseError { .. } => "ParseError",
VfsError::IOError { .. } => "IOError",
VfsError::CapChannelFail { .. } => "CapChannelFail",
VfsError::BadJson { .. } => "NoJson",
VfsError::NotFound { .. } => "NotFound",
VfsError::CreateDirError { .. } => "CreateDirError",
VfsError::Other { .. } => "Other",
}
}
}
impl From<tokio::sync::oneshot::error::RecvError> for VfsError {
fn from(err: tokio::sync::oneshot::error::RecvError) -> Self {
VfsError::CapChannelFail {
error: err.to_string(),
}
}
}
impl From<tokio::sync::mpsc::error::SendError<CapMessage>> for VfsError {
fn from(err: tokio::sync::mpsc::error::SendError<CapMessage>) -> Self {
VfsError::CapChannelFail {
error: err.to_string(),
}
}
#[error("IO error: {0}")]
IOError(String),
#[error("non-file non-dir in zip")]
UnzipError,
}
impl From<std::io::Error> for VfsError {
fn from(err: std::io::Error) -> Self {
VfsError::IOError {
path: "".into(),
error: err.to_string(),
}
}
}
impl std::fmt::Display for VfsAction {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
write!(f, "{:?}", self)
VfsError::IOError(err.to_string())
}
}