feature (plg/extension): document rust extension mechanics

server/pkg/extension/adapter/http.go

@@ -19,6 +19,7 @@ import (
 
 var middlewares = map[string]Middleware{
 	"api_headers":    middleware.ApiHeaders,
+	"index_headers":  middleware.IndexHeaders,
 	"secure_headers": middleware.SecureHeaders,
 	"session_start":  session.Start,
 	"session_try":    session.Try,

server/pkg/extension/adapter/index.go

@@ -19,7 +19,6 @@ func NewInstance(wasm []byte, permissions ...string) (*Instance, error) {
 	}
 	rt, err := runtime.New(wasm, append(opts, runtime.WithExports(func(b *runtime.HostModuleBuilder) {
 		exportShared(b)
-		exportAuthentication(b)
 		exportAuthorisation(b)
 		exportMiddleware(b)
 		exportHttp(b)

server/pkg/extension/adapter/lifecycle.go

@@ -0,0 +1,31 @@
+package adapter
+
+import (
+	"context"
+
+	"github.com/mickael-kerjean/filestash/server/pkg/utils"
+)
+
+func (in *Instance) OnInit() func() {
+	return func() {
+		if err := in.rt.Call(context.Background(), "on_init", appKey{}, nil); err != nil {
+			utils.Log.Error("extension::adapter on_init err=%s", err.Error())
+		}
+	}
+}
+
+func (in *Instance) OnChanges() func() {
+	return func() {
+		if err := in.rt.Call(context.Background(), "on_changes", appKey{}, nil); err != nil {
+			utils.Log.Error("extension::adapter on_changes err=%s", err.Error())
+		}
+	}
+}
+
+func (in *Instance) OnDestroy() func() {
+	return func() {
+		if err := in.rt.Call(context.Background(), "on_destroy", appKey{}, nil); err != nil {
+			utils.Log.Error("extension::adapter on_destroy err=%s", err.Error())
+		}
+	}
+}

server/pkg/extension/adapter/runtime/fs.go

@@ -2,6 +2,7 @@ package runtime
 
 import (
 	"context"
+	"crypto/rand"
 
 	"github.com/tetratelabs/wazero"
 )
@@ -20,6 +21,10 @@ func (f *FSConfig) WithDirMount(host string, guest string) *FSConfig {
 }
 
 func (f *FSConfig) Apply(ctx context.Context, c *config) error {
-	c.mod = wazero.NewModuleConfig().WithFSConfig(f.inner)
+	c.mod = wazero.NewModuleConfig().
+		WithFSConfig(f.inner).
+		WithSysWalltime().
+		WithSysNanotime().
+		WithRandSource(rand.Reader)
 	return nil
 }

server/pkg/extension/adapter/runtime/socket.go

@@ -2,9 +2,10 @@ package runtime
 
 import (
 	"context"
+	"errors"
+	"io"
 	"net"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/tetratelabs/wazero/api"
@@ -57,44 +58,29 @@ func (s *SocketConfig) Apply(ctx context.Context, c *config) error {
 			WithGoModuleFunction(gofn, params, []api.ValueType{i32}).
 			Export(name)
 	}
-	fn("sock_open", api.GoModuleFunc(s.sockOpen), []api.ValueType{i32, i32, i32})
 	fn("sock_connect", api.GoModuleFunc(s.sockConnect), []api.ValueType{i32, i32, i32})
-	fn("sock_send", api.GoModuleFunc(s.sockSend), []api.ValueType{i32, i32, i32, i32, i32})
-	fn("sock_recv", api.GoModuleFunc(s.sockRecv), []api.ValueType{i32, i32, i32, i32, i32, i32})
+	fn("sock_resolve", api.GoModuleFunc(s.sockResolve), []api.ValueType{i32, i32, i32, i32, i32, i32})
+	fn("sock_send", api.GoModuleFunc(s.sockSend), []api.ValueType{i32, i32, i32, i32})
+	fn("sock_recv", api.GoModuleFunc(s.sockRecv), []api.ValueType{i32, i32, i32, i32})
 	fn("sock_shutdown", api.GoModuleFunc(s.sockShutdown), []api.ValueType{i32, i32})
-	fn("sock_setsockopt", api.GoModuleFunc(s.sockSetsockopt), []api.ValueType{i32, i32, i32, i32, i32})
-	fn("sock_getaddrinfo", api.GoModuleFunc(s.sockGetaddrinfo), []api.ValueType{i32, i32, i32, i32, i32, i32, i32, i32})
 	_, err := b.Instantiate(ctx)
 	return err
 }
 
-func (s *SocketConfig) sockOpen(ctx context.Context, mod api.Module, stack []uint64) {
-	s.next += 1
-	s.conns[s.next] = nil
-	mod.Memory().WriteUint32Le(uint32(stack[2]), uint32(s.next))
-	stack[0] = sockSuccess
-}
-
 func (s *SocketConfig) sockConnect(ctx context.Context, mod api.Module, stack []uint64) {
 	var (
-		mem  = mod.Memory()
-		fd   = int32(stack[0])
-		port = uint32(stack[2])
+		mem   = mod.Memory()
+		port  = strconv.Itoa(int(uint32(stack[1])))
+		fdPtr = uint32(stack[2])
 	)
-	if _, ok := s.conns[fd]; !ok {
-		stack[0] = sockErrnoBadf
-		return
-	}
-	bufPtr, _ := mem.ReadUint32Le(uint32(stack[1]))
-	bufLen, _ := mem.ReadUint32Le(uint32(stack[1]) + 4)
-	raw, _ := mem.Read(bufPtr, bufLen)
-	if len(raw) != 4 {
+	raw, ok := mem.Read(uint32(stack[0]), 4)
+	if !ok {
 		stack[0] = sockErrnoInval
 		return
 	}
 	ip := net.IP(raw).String()
-	hostport := net.JoinHostPort(ip, strconv.Itoa(int(port)))
-	if !s.granted[hostport] && !s.granted[net.JoinHostPort(s.resolved[ip], strconv.Itoa(int(port)))] {
+	hostport := net.JoinHostPort(ip, port)
+	if !s.granted[hostport] && !s.granted[net.JoinHostPort(s.resolved[ip], port)] {
 		stack[0] = sockErrnoAcces
 		return
 	}
@@ -103,87 +89,23 @@ func (s *SocketConfig) sockConnect(ctx context.Context, mod api.Module, stack []
 		stack[0] = sockErrnoRefused
 		return
 	}
-	s.conns[fd] = conn
+	s.next += 1
+	s.conns[s.next] = conn
+	mem.WriteUint32Le(fdPtr, uint32(s.next))
 	stack[0] = sockSuccess
 }
 
-func (s *SocketConfig) sockSend(ctx context.Context, mod api.Module, stack []uint64) {
-	var (
-		mem     = mod.Memory()
-		conn    = s.conns[int32(stack[0])]
-		iovsPtr = uint32(stack[1])
-		iovsLen = uint32(stack[2])
-		total   = uint32(0)
-	)
-	if conn == nil {
-		stack[0] = sockErrnoBadf
-		return
-	}
-	for i := uint32(0); i < iovsLen; i++ {
-		bufPtr, _ := mem.ReadUint32Le(iovsPtr + i*8)
-		bufLen, _ := mem.ReadUint32Le(iovsPtr + i*8 + 4)
-		buf, _ := mem.Read(bufPtr, bufLen)
-		n, err := conn.Write(buf)
-		total += uint32(n)
-		if err != nil {
-			stack[0] = sockErrnoIo
-			return
-		}
-	}
-	mem.WriteUint32Le(uint32(stack[4]), total)
-	stack[0] = sockSuccess
-}
-
-func (s *SocketConfig) sockRecv(ctx context.Context, mod api.Module, stack []uint64) {
-	var (
-		mem     = mod.Memory()
-		conn    = s.conns[int32(stack[0])]
-		iovsPtr = uint32(stack[1])
-	)
-	if conn == nil {
-		stack[0] = sockErrnoBadf
-		return
-	}
-	bufPtr, _ := mem.ReadUint32Le(iovsPtr)
-	bufLen, _ := mem.ReadUint32Le(iovsPtr + 4)
-	buf, _ := mem.Read(bufPtr, bufLen)
-	n, err := conn.Read(buf)
-	mem.WriteUint32Le(uint32(stack[4]), uint32(n))
-	mem.WriteUint32Le(uint32(stack[5]), 0)
-	if err != nil && n == 0 {
-		if strings.Contains(err.Error(), "EOF") == false {
-			stack[0] = sockErrnoIo
-			return
-		}
-	}
-	stack[0] = sockSuccess
-}
-
-func (s *SocketConfig) sockShutdown(ctx context.Context, mod api.Module, stack []uint64) {
-	fd := int32(stack[0])
-	if conn := s.conns[fd]; conn != nil {
-		conn.Close()
-	}
-	delete(s.conns, fd)
-	stack[0] = sockSuccess
-}
-
-func (s *SocketConfig) sockSetsockopt(ctx context.Context, mod api.Module, stack []uint64) {
-	stack[0] = sockSuccess
-}
-
-func (s *SocketConfig) sockGetaddrinfo(ctx context.Context, mod api.Module, stack []uint64) {
+func (s *SocketConfig) sockResolve(ctx context.Context, mod api.Module, stack []uint64) {
 	var (
 		mem        = mod.Memory()
-		resPtr     = uint32(stack[5])
-		maxLen     = uint32(stack[6])
-		resLenPtr  = uint32(stack[7])
 		nodeRaw, _ = mem.Read(uint32(stack[0]), uint32(stack[1]))
-		srvRaw, _  = mem.Read(uint32(stack[2]), uint32(stack[3]))
+		port       = strconv.Itoa(int(uint32(stack[2])))
+		outPtr     = uint32(stack[3])
+		max        = uint32(stack[4])
+		countPtr   = uint32(stack[5])
 	)
-	node := strings.TrimRight(string(nodeRaw), "\x00")
-	service := strings.TrimRight(string(srvRaw), "\x00")
-	if !s.granted[net.JoinHostPort(node, service)] {
+	node := string(nodeRaw)
+	if !s.granted[net.JoinHostPort(node, port)] {
 		stack[0] = sockErrnoAcces
 		return
 	}
@@ -198,30 +120,71 @@ func (s *SocketConfig) sockGetaddrinfo(ctx context.Context, mod api.Module, stac
 		}
 		ips = resolved
 	}
-	port, err := strconv.Atoi(service)
-	if err != nil {
-		stack[0] = sockErrnoInval
-		return
-	}
 	count := uint32(0)
-	cur, _ := mem.ReadUint32Le(resPtr)
 	for _, ip := range ips {
 		ip4 := ip.To4()
-		if ip4 == nil || count >= maxLen || cur == 0 {
+		if ip4 == nil || count >= max {
 			continue
 		}
 		s.resolved[ip4.String()] = node
-		mem.WriteUint32Le(cur+8, 16)
-		sockaddrPtr, _ := mem.ReadUint32Le(cur + 12)
-		mem.WriteByte(sockaddrPtr, 1)
-		mem.WriteUint32Le(sockaddrPtr+4, 14)
-		saDataPtr, _ := mem.ReadUint32Le(sockaddrPtr + 8)
-		mem.WriteByte(saDataPtr, byte(port>>8))
-		mem.WriteByte(saDataPtr+1, byte(port))
-		mem.Write(saDataPtr+2, ip4)
-		cur, _ = mem.ReadUint32Le(cur + 24)
+		mem.Write(outPtr+count*4, ip4)
 		count += 1
 	}
-	mem.WriteUint32Le(resLenPtr, count)
+	mem.WriteUint32Le(countPtr, count)
+	stack[0] = sockSuccess
+}
+
+func (s *SocketConfig) sockSend(ctx context.Context, mod api.Module, stack []uint64) {
+	var (
+		mem  = mod.Memory()
+		conn = s.conns[int32(stack[0])]
+	)
+	if conn == nil {
+		stack[0] = sockErrnoBadf
+		return
+	}
+	buf, ok := mem.Read(uint32(stack[1]), uint32(stack[2]))
+	if !ok {
+		stack[0] = sockErrnoInval
+		return
+	}
+	n, err := conn.Write(buf)
+	mem.WriteUint32Le(uint32(stack[3]), uint32(n))
+	if err != nil {
+		stack[0] = sockErrnoIo
+		return
+	}
+	stack[0] = sockSuccess
+}
+
+func (s *SocketConfig) sockRecv(ctx context.Context, mod api.Module, stack []uint64) {
+	var (
+		mem  = mod.Memory()
+		conn = s.conns[int32(stack[0])]
+	)
+	if conn == nil {
+		stack[0] = sockErrnoBadf
+		return
+	}
+	buf, ok := mem.Read(uint32(stack[1]), uint32(stack[2]))
+	if !ok {
+		stack[0] = sockErrnoInval
+		return
+	}
+	n, err := conn.Read(buf)
+	mem.WriteUint32Le(uint32(stack[3]), uint32(n))
+	if err != nil && n == 0 && !errors.Is(err, io.EOF) {
+		stack[0] = sockErrnoIo
+		return
+	}
+	stack[0] = sockSuccess
+}
+
+func (s *SocketConfig) sockShutdown(ctx context.Context, mod api.Module, stack []uint64) {
+	fd := int32(stack[0])
+	if conn := s.conns[fd]; conn != nil {
+		conn.Close()
+	}
+	delete(s.conns, fd)
 	stack[0] = sockSuccess
 }

server/pkg/extension/discovery.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/mickael-kerjean/filestash/server/pkg/config"
 	. "github.com/mickael-kerjean/filestash/server/pkg/env"
 	"github.com/mickael-kerjean/filestash/server/pkg/extension/adapter"
 	. "github.com/mickael-kerjean/filestash/server/pkg/kernel"
@@ -70,6 +71,15 @@ func Discovery() error {
 				if in.Provides("http") {
 					Hooks.Register.HttpEndpoint(in.Http())
 				}
+				if in.Provides("on_init") {
+					Hooks.Register.Onload(in.OnInit())
+				}
+				if in.Provides("on_changes") {
+					config.RegisterChange(in.OnChanges())
+				}
+				if in.Provides("on_destroy") {
+					Hooks.Register.OnQuit(in.OnDestroy())
+				}
 			case "xdg-open": // noop
 			default:
 				return fmt.Errorf("%w: %s", ErrNotImplemented, impl.Modules[i]["type"])

server/pkg/extension/sdk/rust/src/app.rs

@@ -85,6 +85,33 @@ macro_rules! register {
         }
     };
 
+    (@capability $anchor:ident, $app:ident, OnInit) => {
+        #[no_mangle]
+        pub extern "C" fn capability_on_init() {}
+        #[no_mangle]
+        pub extern "C" fn on_init() {
+            $anchor.with(|app| <$app as $crate::OnInit>::on_init(app));
+        }
+    };
+
+    (@capability $anchor:ident, $app:ident, OnChanges) => {
+        #[no_mangle]
+        pub extern "C" fn capability_on_changes() {}
+        #[no_mangle]
+        pub extern "C" fn on_changes() {
+            $anchor.with(|app| <$app as $crate::OnChanges>::on_changes(app));
+        }
+    };
+
+    (@capability $anchor:ident, $app:ident, OnDestroy) => {
+        #[no_mangle]
+        pub extern "C" fn capability_on_destroy() {}
+        #[no_mangle]
+        pub extern "C" fn on_destroy() {
+            $anchor.with(|app| <$app as $crate::OnDestroy>::on_destroy(app));
+        }
+    };
+
     (@capability $anchor:ident, $app:ident, Middleware) => {
         #[no_mangle]
         pub extern "C" fn capability_middleware() {}

server/pkg/extension/sdk/rust/src/lib.rs

@@ -3,6 +3,7 @@ pub mod authorisation;
 mod context;
 mod ffi;
 mod http;
+mod lifecycle;
 pub mod logger;
 mod middleware;
 mod request;
@@ -12,6 +13,7 @@ pub use app::App;
 pub use authorisation::{Authorisation, Decision};
 pub use context::Context;
 pub use http::{Handler, Http, Router};
+pub use lifecycle::{OnChanges, OnDestroy, OnInit};
 pub use middleware::{Middleware, Next};
 pub use request::Request;
 pub use response::Response;

server/pkg/extension/sdk/rust/src/lifecycle.rs

@@ -0,0 +1,11 @@
+pub trait OnInit {
+    fn on_init(&self);
+}
+
+pub trait OnChanges {
+    fn on_changes(&self);
+}
+
+pub trait OnDestroy {
+    fn on_destroy(&self);
+}

server/pkg/extension/sdk/rust/src/logger.rs

@@ -19,5 +19,5 @@ static LOGGER: Logger = Logger;
 
 pub fn init() {
     let _ = log::set_logger(&LOGGER);
-    log::set_max_level(log::LevelFilter::Trace);
+    log::set_max_level(log::LevelFilter::Info);
 }

server/plugin/plg_runtime_cookbook/.gitignore

@@ -0,0 +1,3 @@
+target
+Cargo.lock
+*.wasm

server/plugin/plg_runtime_cookbook/Makefile

@@ -0,0 +1,29 @@
+RELEASE = target/wasm32-wasip1/release
+
+all:
+	$(MAKE) -C http
+	$(MAKE) -C onInit
+	$(MAKE) -C onChange
+	$(MAKE) -C onQuit
+	$(MAKE) -C socket
+	$(MAKE) -C filesystem
+	zip -j example.zip manifest.json \
+		http/$(RELEASE)/endpoint.wasm \
+		onInit/$(RELEASE)/oninit.wasm \
+		onChange/$(RELEASE)/onchange.wasm \
+		onQuit/$(RELEASE)/onquit.wasm \
+		socket/$(RELEASE)/socket.wasm \
+		filesystem/$(RELEASE)/filesystem.wasm
+	mv example.zip ../../../dist/data/state/plugins/
+
+init:
+	rustup target add wasm32-wasip1
+	$(MAKE) -C socket init
+
+clean:
+	$(MAKE) -C http clean
+	$(MAKE) -C onInit clean
+	$(MAKE) -C onChange clean
+	$(MAKE) -C onQuit clean
+	$(MAKE) -C socket clean
+	$(MAKE) -C filesystem clean

server/plugin/plg_runtime_cookbook/README.md

@@ -0,0 +1 @@
+This is just some example of runtime plugins for Filestash

server/plugin/plg_runtime_cookbook/filesystem/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "filesystem"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/filesystem/Makefile

@@ -0,0 +1,5 @@
+all:
+	cargo build --release --target wasm32-wasip1
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/filesystem/lib.rs

@@ -0,0 +1,35 @@
+use filestash::*;
+use std::fs::OpenOptions;
+use std::io::Write;
+
+pub struct Plugin;
+
+impl App for Plugin {
+    fn new() -> Self {
+        Plugin
+    }
+}
+
+impl OnInit for Plugin {
+    fn on_init(&self) {
+        append("the server has started\n");
+    }
+}
+
+impl OnDestroy for Plugin {
+    fn on_destroy(&self) {
+        append("the server has stopped\n");
+    }
+}
+
+fn append(line: &str) {
+    let mut file = OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open("/{{ .TMP_PATH }}/example.log")
+        .unwrap();
+    file.write_all(line.as_bytes()).unwrap();
+    log::info!("[runtime::plugin::filesystem] appended to example.log: {}", line.trim());
+}
+
+register!(Plugin: OnInit + OnDestroy);

server/plugin/plg_runtime_cookbook/http/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "endpoint"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/http/Makefile

@@ -0,0 +1,5 @@
+all:
+	cargo build --release --target wasm32-wasip1
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/http/lib.rs

@@ -0,0 +1,25 @@
+use filestash::*;
+
+pub struct Page;
+
+impl App for Page {
+    fn new() -> Self {
+        Page
+    }
+}
+
+impl Http for Page {
+    fn routes(r: &mut Router<Self>) {
+        r.get("/api/example", &["index_headers"], Page::index);
+    }
+}
+
+impl Page {
+    fn index(&self, _req: &Request, res: &mut Response) {
+        res.header("Content-Type", "text/html; charset=utf-8")
+            .status(200)
+            .write(b"<!DOCTYPE html><h1>Hello from a Filestash plugin</h1>");
+    }
+}
+
+register!(Page: Http);

server/plugin/plg_runtime_cookbook/manifest.json

@@ -0,0 +1,16 @@
+{
+    "author": "Filestash Pty Ltd",
+    "version": "v0.0",
+    "modules": [
+        { "type": "abi", "entrypoint": "endpoint.wasm" },
+        { "type": "abi", "entrypoint": "oninit.wasm" },
+        { "type": "abi", "entrypoint": "onchange.wasm" },
+        { "type": "abi", "entrypoint": "onquit.wasm" },
+        { "type": "abi", "entrypoint": "socket.wasm" },
+        { "type": "abi", "entrypoint": "filesystem.wasm" }
+    ],
+    "permissions": [
+        "tcp://demo.filestash.app:443",
+        "file:///{{ .TMP_PATH }}"
+    ]
+}

server/plugin/plg_runtime_cookbook/onChange/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "onchange"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/onChange/Makefile

@@ -0,0 +1,5 @@
+all:
+	cargo build --release --target wasm32-wasip1
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/onChange/lib.rs

@@ -0,0 +1,17 @@
+use filestash::*;
+
+pub struct Plugin;
+
+impl App for Plugin {
+    fn new() -> Self {
+        Plugin
+    }
+}
+
+impl OnChanges for Plugin {
+    fn on_changes(&self) {
+        log::info!("[runtime::plugin::onchange] config has changed");
+    }
+}
+
+register!(Plugin: OnChanges);

server/plugin/plg_runtime_cookbook/onInit/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "oninit"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/onInit/Makefile

@@ -0,0 +1,5 @@
+all:
+	cargo build --release --target wasm32-wasip1
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/onInit/lib.rs

@@ -0,0 +1,17 @@
+use filestash::*;
+
+pub struct Plugin;
+
+impl App for Plugin {
+    fn new() -> Self {
+        Plugin
+    }
+}
+
+impl OnInit for Plugin {
+    fn on_init(&self) {
+        log::info!("[runtime::plugin::oninit] server is ready");
+    }
+}
+
+register!(Plugin: OnInit);

server/plugin/plg_runtime_cookbook/onQuit/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "onquit"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/onQuit/Makefile

@@ -0,0 +1,5 @@
+all:
+	cargo build --release --target wasm32-wasip1
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/onQuit/lib.rs

@@ -0,0 +1,17 @@
+use filestash::*;
+
+pub struct Plugin;
+
+impl App for Plugin {
+    fn new() -> Self {
+        Plugin
+    }
+}
+
+impl OnDestroy for Plugin {
+    fn on_destroy(&self) {
+        log::info!("[runtime::plugin::onquit] server is shutting down");
+    }
+}
+
+register!(Plugin: OnDestroy);

server/plugin/plg_runtime_cookbook/socket/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "socket"
+edition = "2021"
+
+[lib]
+path = "lib.rs"
+crate-type = ["cdylib"]
+
+[dependencies]
+filestash = { path = "../../../pkg/extension/sdk/rust" }
+ureq = { version = "3", default-features = false, features = ["rustls"] }
+
+[profile.release]
+opt-level = "z"
+lto = true
+strip = true

server/plugin/plg_runtime_cookbook/socket/Makefile

@@ -0,0 +1,9 @@
+all:
+	cargo +nightly build --release -Zbuild-std=std,panic_abort --target wasm32-wasip1
+
+init:
+	rustup toolchain install nightly --component rust-src --target wasm32-wasip1
+	cd "$$(rustc +nightly --print sysroot)/lib/rustlib/src/rust" && patch -p1 --forward -r - < $(CURDIR)/enable-network.patch || true
+
+clean:
+	rm -rf target

server/plugin/plg_runtime_cookbook/socket/enable-network.patch

@@ -0,0 +1,251 @@
+--- a/library/std/src/sys/net/connection/wasip1.rs
++++ b/library/std/src/sys/net/connection/wasip1.rs
+@@ -2,7 +2,8 @@
+ 
+ use crate::fmt;
+ use crate::io::{self, BorrowedCursor, IoSlice, IoSliceMut};
+-use crate::net::{Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, ToSocketAddrs};
++use crate::net::{Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, SocketAddrV4, ToSocketAddrs};
++use crate::sync::atomic::{AtomicBool, Ordering};
+ use crate::os::wasi::io::{AsFd, AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, RawFd};
+ use crate::sys::fd::FileDesc;
+ use crate::sys::{AsInner, FromInner, IntoInner, err2io, unsupported};
+@@ -12,6 +13,7 @@
+ 
+ pub struct TcpStream {
+     inner: Socket,
++    nonblocking: AtomicBool,
+ }
+ 
+ impl AsInner<FileDesc> for Socket {
+@@ -58,29 +60,76 @@
+     }
+ }
+ 
++mod sock {
++    #[link(wasm_import_module = "wasi_snapshot_preview1")]
++    unsafe extern "C" {
++        pub fn sock_connect(ip: *const u8, port: u32, fd_out: *mut u32) -> u32;
++        pub fn sock_resolve(
++            node: *const u8,
++            node_len: u32,
++            port: u32,
++            out: *mut u8,
++            max: u32,
++            count_out: *mut u32,
++        ) -> u32;
++        pub fn sock_send(fd: u32, buf: *const u8, len: u32, n_out: *mut u32) -> u32;
++        pub fn sock_recv(fd: u32, buf: *mut u8, len: u32, n_out: *mut u32) -> u32;
++    }
++}
++
++fn cvt_sock(errno: u32) -> io::Result<()> {
++    match errno {
++        0 => Ok(()),
++        2 => Err(io::Error::new(io::ErrorKind::PermissionDenied, "host not allowed")),
++        8 => Err(io::Error::new(io::ErrorKind::NotConnected, "bad socket descriptor")),
++        14 => Err(io::Error::new(io::ErrorKind::ConnectionRefused, "connection refused")),
++        28 => Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid argument")),
++        _ => Err(io::Error::new(io::ErrorKind::Uncategorized, "socket io error")),
++    }
++}
++
+ impl TcpStream {
+-    pub fn connect<A: ToSocketAddrs>(_: A) -> io::Result<TcpStream> {
+-        unsupported()
++    pub fn connect<A: ToSocketAddrs>(addr: A) -> io::Result<TcpStream> {
++        let mut last_err = None;
++        for addr in addr.to_socket_addrs()? {
++            match Self::connect_one(&addr) {
++                Ok(stream) => return Ok(stream),
++                Err(e) => last_err = Some(e),
++            }
++        }
++        Err(last_err.unwrap_or_else(|| {
++            io::Error::new(io::ErrorKind::InvalidInput, "could not resolve to any addresses")
++        }))
+     }
+ 
+-    pub fn connect_timeout(_: &SocketAddr, _: Duration) -> io::Result<TcpStream> {
+-        unsupported()
++    fn connect_one(addr: &SocketAddr) -> io::Result<TcpStream> {
++        let SocketAddr::V4(v4) = addr else {
++            return Err(io::Error::new(io::ErrorKind::Unsupported, "only ipv4 is supported"));
++        };
++        let ip = v4.ip().octets();
++        let mut fd: u32 = 0;
++        cvt_sock(unsafe { sock::sock_connect(ip.as_ptr(), v4.port() as u32, &mut fd) })?;
++        Ok(TcpStream::from_inner(unsafe { Socket::from_raw_fd(fd as RawFd) }))
++    }
++
++    pub fn connect_timeout(addr: &SocketAddr, _: Duration) -> io::Result<TcpStream> {
++        Self::connect_one(addr)
+     }
+ 
+     pub fn set_read_timeout(&self, _: Option<Duration>) -> io::Result<()> {
+-        unsupported()
++        Ok(())
+     }
+ 
+     pub fn set_write_timeout(&self, _: Option<Duration>) -> io::Result<()> {
+-        unsupported()
++        Ok(())
+     }
+ 
+     pub fn read_timeout(&self) -> io::Result<Option<Duration>> {
+-        unsupported()
++        Ok(None)
+     }
+ 
+     pub fn write_timeout(&self) -> io::Result<Option<Duration>> {
+-        unsupported()
++        Ok(None)
+     }
+ 
+     pub fn peek(&self, _: &mut [u8]) -> io::Result<usize> {
+@@ -88,31 +137,52 @@
+     }
+ 
+     pub fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
+-        self.read_vectored(&mut [IoSliceMut::new(buf)])
++        if self.nonblocking.load(Ordering::Relaxed) {
++            return Err(io::Error::from(io::ErrorKind::WouldBlock));
++        }
++        let mut n: u32 = 0;
++        cvt_sock(unsafe {
++            sock::sock_recv(self.inner.as_raw_fd() as u32, buf.as_mut_ptr(), buf.len() as u32, &mut n)
++        })?;
++        Ok(n as usize)
+     }
+ 
+-    pub fn read_buf(&self, buf: BorrowedCursor<'_, u8>) -> io::Result<()> {
+-        self.socket().as_inner().read_buf(buf)
++    pub fn read_buf(&self, mut buf: BorrowedCursor<'_, u8>) -> io::Result<()> {
++        let mut tmp = [0u8; 8192];
++        let max = tmp.len().min(buf.capacity());
++        let n = self.read(&mut tmp[..max])?;
++        buf.append(&tmp[..n]);
++        Ok(())
+     }
+ 
+     pub fn read_vectored(&self, bufs: &mut [IoSliceMut<'_>]) -> io::Result<usize> {
+-        self.socket().as_inner().read_vectored(bufs)
++        match bufs.iter_mut().find(|b| !b.is_empty()) {
++            Some(buf) => self.read(&mut buf[..]),
++            None => Ok(0),
++        }
+     }
+ 
+     pub fn is_read_vectored(&self) -> bool {
+-        true
++        false
+     }
+ 
+     pub fn write(&self, buf: &[u8]) -> io::Result<usize> {
+-        self.write_vectored(&[IoSlice::new(buf)])
++        let mut n: u32 = 0;
++        cvt_sock(unsafe {
++            sock::sock_send(self.inner.as_raw_fd() as u32, buf.as_ptr(), buf.len() as u32, &mut n)
++        })?;
++        Ok(n as usize)
+     }
+ 
+     pub fn write_vectored(&self, bufs: &[IoSlice<'_>]) -> io::Result<usize> {
+-        self.socket().as_inner().write_vectored(bufs)
++        match bufs.iter().find(|b| !b.is_empty()) {
++            Some(buf) => self.write(&buf[..]),
++            None => Ok(0),
++        }
+     }
+ 
+     pub fn is_write_vectored(&self) -> bool {
+-        true
++        false
+     }
+ 
+     pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+@@ -154,11 +224,11 @@
+     }
+ 
+     pub fn set_nodelay(&self, _: bool) -> io::Result<()> {
+-        unsupported()
++        Ok(())
+     }
+ 
+     pub fn nodelay(&self) -> io::Result<bool> {
+-        unsupported()
++        Ok(true)
+     }
+ 
+     pub fn set_ttl(&self, _: u32) -> io::Result<()> {
+@@ -174,23 +244,8 @@
+     }
+ 
+     pub fn set_nonblocking(&self, state: bool) -> io::Result<()> {
+-        let fdstat = unsafe {
+-            wasip1::fd_fdstat_get(self.socket().as_inner().as_raw_fd() as wasip1::Fd)
+-                .map_err(err2io)?
+-        };
+-
+-        let mut flags = fdstat.fs_flags;
+-
+-        if state {
+-            flags |= wasip1::FDFLAGS_NONBLOCK;
+-        } else {
+-            flags &= !wasip1::FDFLAGS_NONBLOCK;
+-        }
+-
+-        unsafe {
+-            wasip1::fd_fdstat_set_flags(self.socket().as_inner().as_raw_fd() as wasip1::Fd, flags)
+-                .map_err(err2io)
+-        }
++        self.nonblocking.store(state, Ordering::Relaxed);
++        Ok(())
+     }
+ 
+     #[inline]
+@@ -205,7 +260,7 @@
+ 
+ impl FromInner<Socket> for TcpStream {
+     fn from_inner(socket: Socket) -> TcpStream {
+-        TcpStream { inner: socket }
++        TcpStream { inner: socket, nonblocking: AtomicBool::new(false) }
+     }
+ }
+ 
+@@ -484,15 +539,31 @@
+     }
+ }
+ 
+-pub struct LookupHost(!);
++pub struct LookupHost {
++    ips: [u8; 32],
++    count: usize,
++    pos: usize,
++    port: u16,
++}
+ 
+ impl Iterator for LookupHost {
+     type Item = SocketAddr;
+     fn next(&mut self) -> Option<SocketAddr> {
+-        self.0
++        if self.pos >= self.count {
++            return None;
++        }
++        let i = self.pos * 4;
++        self.pos += 1;
++        let ip = Ipv4Addr::new(self.ips[i], self.ips[i + 1], self.ips[i + 2], self.ips[i + 3]);
++        Some(SocketAddr::V4(SocketAddrV4::new(ip, self.port)))
+     }
+ }
+ 
+-pub fn lookup_host(_host: &str, _port: u16) -> io::Result<LookupHost> {
+-    unsupported()
++pub fn lookup_host(host: &str, port: u16) -> io::Result<LookupHost> {
++    let mut ips = [0u8; 32];
++    let mut count: u32 = 0;
++    cvt_sock(unsafe {
++        sock::sock_resolve(host.as_ptr(), host.len() as u32, port as u32, ips.as_mut_ptr(), 8, &mut count)
++    })?;
++    Ok(LookupHost { ips, count: (count as usize).min(8), pos: 0, port })
+ }

server/plugin/plg_runtime_cookbook/socket/lib.rs

@@ -0,0 +1,23 @@
+use filestash::*;
+
+pub struct Plugin;
+
+impl App for Plugin {
+    fn new() -> Self {
+        Plugin
+    }
+}
+
+impl OnInit for Plugin {
+    fn on_init(&self) {
+        match ureq::get("https://demo.filestash.app/healthz").call() {
+            Ok(mut r) => match r.body_mut().read_to_string() {
+                Ok(body) => log::info!("[runtime::plugin::socket] healthz {}", body),
+                Err(e) => log::error!("[runtime::plugin::socket] healthz {}", e),
+            },
+            Err(e) => log::error!("[runtime::plugin::socket] healthz {}", e),
+        }
+    }
+}
+
+register!(Plugin: OnInit);