mirror of
https://github.com/DeBrosOfficial/orama.git
synced 2026-06-16 22:54:12 +00:00
Changes
This commit is contained in:
parent
0d352d0b42
commit
325a2471c7
@ -735,6 +735,7 @@ func (e *Engine) registerHostModule(ctx context.Context) error {
|
||||
NewFunctionBuilder().WithFunc(e.hCacheIncr).Export("cache_incr").
|
||||
NewFunctionBuilder().WithFunc(e.hCacheIncrBy).Export("cache_incr_by").
|
||||
NewFunctionBuilder().WithFunc(e.hHTTPFetch).Export("http_fetch").
|
||||
NewFunctionBuilder().WithFunc(e.hAnyoneFetch).Export("anyone_fetch").
|
||||
NewFunctionBuilder().WithFunc(e.hPubSubPublish).Export("pubsub_publish").
|
||||
NewFunctionBuilder().WithFunc(e.hPubSubPublishBatch).Export("pubsub_publish_batch").
|
||||
NewFunctionBuilder().WithFunc(e.hPushSend).Export("push_send").
|
||||
@ -940,6 +941,43 @@ func (e *Engine) hHTTPFetch(ctx context.Context, mod api.Module, methodPtr, meth
|
||||
return e.executor.WriteToGuest(ctx, mod, resp)
|
||||
}
|
||||
|
||||
// hAnyoneFetch is the WASM-callable wrapper for AnyoneFetch — feat-11.
|
||||
// Identical ABI to hHTTPFetch (method, url, headers JSON, body), routes
|
||||
// through the Anyone SOCKS5 proxy. Returns packed (ptr<<32 | len) to the
|
||||
// JSON response envelope, or 0 on a setup error (the typed
|
||||
// proxy-unavailable / transport-error cases come back inside the
|
||||
// envelope with status 0, NOT as a 0 return).
|
||||
func (e *Engine) hAnyoneFetch(ctx context.Context, mod api.Module, methodPtr, methodLen, urlPtr, urlLen, headersPtr, headersLen, bodyPtr, bodyLen uint32) uint64 {
|
||||
method, ok := e.executor.ReadFromGuest(mod, methodPtr, methodLen)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
u, ok := e.executor.ReadFromGuest(mod, urlPtr, urlLen)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
|
||||
var headers map[string]string
|
||||
if headersLen > 0 {
|
||||
if err := e.executor.UnmarshalJSONFromGuest(mod, headersPtr, headersLen, &headers); err != nil {
|
||||
e.logger.Error("failed to unmarshal anyone_fetch headers", zap.Error(err))
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
body, ok := e.executor.ReadFromGuest(mod, bodyPtr, bodyLen)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
|
||||
resp, err := e.hostServices.AnyoneFetch(ctx, string(method), string(u), headers, body)
|
||||
if err != nil {
|
||||
e.logger.Error("host function anyone_fetch failed", zap.Error(err), zap.String("url", string(u)))
|
||||
return 0
|
||||
}
|
||||
return e.executor.WriteToGuest(ctx, mod, resp)
|
||||
}
|
||||
|
||||
func (e *Engine) hPubSubPublish(ctx context.Context, mod api.Module, topicPtr, topicLen, dataPtr, dataLen uint32) uint32 {
|
||||
topic, ok := e.executor.ReadFromGuest(mod, topicPtr, topicLen)
|
||||
if !ok {
|
||||
|
||||
@ -150,6 +150,10 @@ func (m *mockHostServices) HTTPFetch(ctx context.Context, method, url string, he
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockHostServices) AnyoneFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *mockHostServices) GetEnv(ctx context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
129
core/pkg/serverless/hostfunctions/anyone_fetch_test.go
Normal file
129
core/pkg/serverless/hostfunctions/anyone_fetch_test.go
Normal file
@ -0,0 +1,129 @@
|
||||
package hostfunctions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// feat-11 — AnyoneFetch (Anyone-routed outbound HTTP for serverless fns).
|
||||
//
|
||||
// The privacy contract is the part that matters: there must be NO silent
|
||||
// fallback to the direct path when Anyone routing is unavailable. A
|
||||
// privacy regression has to fail loudly (typed error), never degrade to
|
||||
// a direct send that leaks the gateway↔upstream metadata trail the
|
||||
// caller was trying to hide.
|
||||
|
||||
func TestAnyoneFetch_nilClientReturnsTypedErrorNotDirectSend(t *testing.T) {
|
||||
// The critical guarantee. When Anyone routing is disabled on this
|
||||
// gateway, anyoneHTTPClient is nil. AnyoneFetch MUST return the
|
||||
// typed {error, status:0, proxy:"anyone"} envelope — NOT silently
|
||||
// dial direct. If this regresses, every wallet-RPC call AnChat
|
||||
// routes through anyone_fetch would leak over the gateway's direct
|
||||
// egress without anyone noticing.
|
||||
h := &HostFunctions{
|
||||
logger: zap.NewNop(),
|
||||
// anyoneHTTPClient intentionally nil (Anyone disabled)
|
||||
}
|
||||
|
||||
raw, err := h.AnyoneFetch(context.Background(), "GET", "https://rpc.example.com", nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("AnyoneFetch returned Go error; want typed envelope: %v", err)
|
||||
}
|
||||
var env map[string]interface{}
|
||||
if e := json.Unmarshal(raw, &env); e != nil {
|
||||
t.Fatalf("unmarshal envelope: %v", e)
|
||||
}
|
||||
if env["status"] != float64(0) {
|
||||
t.Errorf("status = %v; want 0 (transport/setup failure marker)", env["status"])
|
||||
}
|
||||
if env["proxy"] != "anyone" {
|
||||
t.Errorf("proxy = %v; want \"anyone\" (so caller can distinguish anyone-path failure)", env["proxy"])
|
||||
}
|
||||
errStr, _ := env["error"].(string)
|
||||
if errStr == "" {
|
||||
t.Error("error field empty; want an actionable 'anyone routing not available' message")
|
||||
}
|
||||
// The envelope must NOT contain a body — a nil client means we never
|
||||
// made a request, so there's no upstream response. Presence of a
|
||||
// body here would imply a direct send happened.
|
||||
if _, hasBody := env["body"]; hasBody {
|
||||
t.Error("PRIVACY REGRESSION: envelope has a body — a request was made despite nil anyone client (silent direct fallback?)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnyoneFetch_routesThroughConfiguredClient(t *testing.T) {
|
||||
// When an Anyone client IS configured, AnyoneFetch uses it (here a
|
||||
// stand-in pointing at a local test server — the SOCKS dialer is
|
||||
// exercised by the anyoneproxy package's own tests; here we verify
|
||||
// AnyoneFetch threads the request through whatever client it was
|
||||
// given and shapes the response envelope correctly).
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("X-Test", "ok")
|
||||
w.WriteHeader(200)
|
||||
_, _ = w.Write([]byte(`{"jsonrpc":"2.0","result":"0x1"}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
h := &HostFunctions{
|
||||
logger: zap.NewNop(),
|
||||
anyoneHTTPClient: srv.Client(), // stand-in for the SOCKS-routed client
|
||||
}
|
||||
|
||||
raw, err := h.AnyoneFetch(context.Background(), "POST", srv.URL,
|
||||
map[string]string{"Content-Type": "application/json"},
|
||||
[]byte(`{"method":"getBalance"}`))
|
||||
if err != nil {
|
||||
t.Fatalf("AnyoneFetch: %v", err)
|
||||
}
|
||||
var env map[string]interface{}
|
||||
_ = json.Unmarshal(raw, &env)
|
||||
|
||||
if env["status"] != float64(200) {
|
||||
t.Errorf("status = %v; want 200", env["status"])
|
||||
}
|
||||
body, _ := env["body"].(string)
|
||||
if body != `{"jsonrpc":"2.0","result":"0x1"}` {
|
||||
t.Errorf("body = %q; want the upstream JSON-RPC response", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnyoneFetch_andHTTPFetch_shareEnvelopeShape(t *testing.T) {
|
||||
// Both fetch variants must produce the SAME envelope shape
|
||||
// (status/headers/body) so a function can swap http_fetch ↔
|
||||
// anyone_fetch without changing its response parsing. Pin it by
|
||||
// running the same upstream through both and comparing keys.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte("hello"))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
h := &HostFunctions{
|
||||
logger: zap.NewNop(),
|
||||
httpClient: srv.Client(),
|
||||
anyoneHTTPClient: srv.Client(),
|
||||
}
|
||||
|
||||
directRaw, _ := h.HTTPFetch(context.Background(), "GET", srv.URL, nil, nil)
|
||||
anyoneRaw, _ := h.AnyoneFetch(context.Background(), "GET", srv.URL, nil, nil)
|
||||
|
||||
var d, a map[string]interface{}
|
||||
_ = json.Unmarshal(directRaw, &d)
|
||||
_ = json.Unmarshal(anyoneRaw, &a)
|
||||
|
||||
for _, k := range []string{"status", "headers", "body"} {
|
||||
if _, ok := d[k]; !ok {
|
||||
t.Errorf("http_fetch envelope missing %q", k)
|
||||
}
|
||||
if _, ok := a[k]; !ok {
|
||||
t.Errorf("anyone_fetch envelope missing %q (must match http_fetch shape)", k)
|
||||
}
|
||||
}
|
||||
if d["body"] != a["body"] || d["body"] != "hello" {
|
||||
t.Errorf("bodies differ: direct=%v anyone=%v", d["body"], a["body"])
|
||||
}
|
||||
}
|
||||
@ -1,8 +1,10 @@
|
||||
package hostfunctions
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/DeBrosOfficial/network/pkg/anyoneproxy"
|
||||
"github.com/DeBrosOfficial/network/pkg/ipfs"
|
||||
"github.com/DeBrosOfficial/network/pkg/pubsub"
|
||||
"github.com/DeBrosOfficial/network/pkg/push"
|
||||
@ -42,6 +44,19 @@ func NewHostFunctions(
|
||||
httpTimeout = 30 * time.Second
|
||||
}
|
||||
|
||||
// Build the Anyone-routed HTTP client only when Anyone routing is
|
||||
// enabled on this gateway (feat-11). When disabled, leave it nil so
|
||||
// AnyoneFetch returns a typed error instead of silently using the
|
||||
// direct path. anyoneproxy.NewHTTPClient() returns a fresh client
|
||||
// with a SOCKS transport when enabled — safe to set Timeout on it
|
||||
// (when disabled it returns the shared http.DefaultClient, which we
|
||||
// must NOT mutate; the Enabled() guard ensures we never reach that).
|
||||
var anyoneHTTPClient *http.Client
|
||||
if anyoneproxy.Enabled() {
|
||||
anyoneHTTPClient = anyoneproxy.NewHTTPClient()
|
||||
anyoneHTTPClient.Timeout = httpTimeout
|
||||
}
|
||||
|
||||
return &HostFunctions{
|
||||
db: db,
|
||||
cacheClient: cacheClient,
|
||||
@ -53,6 +68,7 @@ func NewHostFunctions(
|
||||
pushDispatcher: pushDispatcher,
|
||||
pushManager: pushManager,
|
||||
wsBridge: wsBridge,
|
||||
anyoneHTTPClient: anyoneHTTPClient,
|
||||
turnDomain: cfg.TURNDomain,
|
||||
turnSecret: cfg.TURNSecret,
|
||||
stealthCDNDomain: cfg.StealthCDNDomain,
|
||||
|
||||
@ -12,8 +12,46 @@ import (
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// HTTPFetch makes an outbound HTTP request.
|
||||
// HTTPFetch makes an outbound HTTP request directly from the gateway.
|
||||
func (h *HostFunctions) HTTPFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error) {
|
||||
return h.doFetch(ctx, "http_fetch", h.httpClient, method, url, headers, body)
|
||||
}
|
||||
|
||||
// AnyoneFetch makes an outbound HTTP request routed through the Anyone
|
||||
// (ANyONe protocol) SOCKS5 proxy, so the third-party endpoint sees an
|
||||
// Anyone exit IP instead of the gateway IP and the gateway can't
|
||||
// correlate (function → external request) traffic by source IP.
|
||||
// Feat-11 — server-side analog of anchat's client-side proxyClient.
|
||||
//
|
||||
// Privacy guarantee: there is NO silent fallback to direct. If Anyone
|
||||
// routing isn't available on this gateway (operator disabled it via
|
||||
// --disable-anonrc / ANYONE_DISABLE=1, so h.anyoneHTTPClient is nil),
|
||||
// this returns a typed error rather than leaking the request over the
|
||||
// direct path. If the Anyone daemon is configured-but-down, the SOCKS
|
||||
// dial to localhost:9050 fails and surfaces as a transport error — also
|
||||
// never a direct send. This is the explicit ask in feat-11: a privacy
|
||||
// regression must fail loudly, not degrade silently.
|
||||
func (h *HostFunctions) AnyoneFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error) {
|
||||
if h.anyoneHTTPClient == nil {
|
||||
// Anyone routing not enabled on this gateway. Return the typed
|
||||
// error envelope (status 0) rather than dialing direct — the
|
||||
// caller explicitly asked for anonymized egress and we must not
|
||||
// silently downgrade it.
|
||||
errorResp := map[string]interface{}{
|
||||
"error": "anyone routing not available on this gateway (disabled by operator)",
|
||||
"status": 0,
|
||||
"proxy": "anyone",
|
||||
}
|
||||
return json.Marshal(errorResp)
|
||||
}
|
||||
return h.doFetch(ctx, "anyone_fetch", h.anyoneHTTPClient, method, url, headers, body)
|
||||
}
|
||||
|
||||
// doFetch is the shared request/response machinery for HTTPFetch and
|
||||
// AnyoneFetch — identical except for which *http.Client (direct vs
|
||||
// SOCKS-routed) does the dialing and the function name used in logs +
|
||||
// HostFunctionError.
|
||||
func (h *HostFunctions) doFetch(ctx context.Context, fnName string, client *http.Client, method, url string, headers map[string]string, body []byte) ([]byte, error) {
|
||||
var bodyReader io.Reader
|
||||
if len(body) > 0 {
|
||||
bodyReader = bytes.NewReader(body)
|
||||
@ -21,7 +59,7 @@ func (h *HostFunctions) HTTPFetch(ctx context.Context, method, url string, heade
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, method, url, bodyReader)
|
||||
if err != nil {
|
||||
h.logger.Error("http_fetch request creation error", zap.Error(err), zap.String("url", url))
|
||||
h.logger.Error(fnName+" request creation error", zap.Error(err), zap.String("url", url))
|
||||
errorResp := map[string]interface{}{
|
||||
"error": "failed to create request: " + err.Error(),
|
||||
"status": 0,
|
||||
@ -33,9 +71,9 @@ func (h *HostFunctions) HTTPFetch(ctx context.Context, method, url string, heade
|
||||
req.Header.Set(key, value)
|
||||
}
|
||||
|
||||
resp, err := h.httpClient.Do(req)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
h.logger.Error("http_fetch transport error", zap.Error(err), zap.String("url", url))
|
||||
h.logger.Error(fnName+" transport error", zap.Error(err), zap.String("url", url))
|
||||
errorResp := map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
"status": 0, // Transport error
|
||||
@ -46,7 +84,7 @@ func (h *HostFunctions) HTTPFetch(ctx context.Context, method, url string, heade
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
h.logger.Error("http_fetch response read error", zap.Error(err), zap.String("url", url))
|
||||
h.logger.Error(fnName+" response read error", zap.Error(err), zap.String("url", url))
|
||||
errorResp := map[string]interface{}{
|
||||
"error": "failed to read response: " + err.Error(),
|
||||
"status": resp.StatusCode,
|
||||
@ -63,7 +101,7 @@ func (h *HostFunctions) HTTPFetch(ctx context.Context, method, url string, heade
|
||||
|
||||
data, err := json.Marshal(response)
|
||||
if err != nil {
|
||||
return nil, &serverless.HostFunctionError{Function: "http_fetch", Cause: fmt.Errorf("failed to marshal response: %w", err)}
|
||||
return nil, &serverless.HostFunctionError{Function: fnName, Cause: fmt.Errorf("failed to marshal response: %w", err)}
|
||||
}
|
||||
|
||||
return data, nil
|
||||
|
||||
@ -42,7 +42,13 @@ type HostFunctions struct {
|
||||
wsManager serverless.WebSocketManager
|
||||
secrets serverless.SecretsManager
|
||||
httpClient *http.Client
|
||||
logger *zap.Logger
|
||||
// anyoneHTTPClient routes outbound requests through the Anyone SOCKS5
|
||||
// proxy (feat-11). nil when Anyone routing is disabled on this
|
||||
// gateway — AnyoneFetch returns a typed error in that case rather
|
||||
// than falling back to the direct httpClient (no silent privacy
|
||||
// regression).
|
||||
anyoneHTTPClient *http.Client
|
||||
logger *zap.Logger
|
||||
|
||||
// pushDispatcher (legacy) and pushManager (per-namespace, bug #220
|
||||
// follow-up) provide push send-paths. When pushManager is set, PushSend
|
||||
|
||||
@ -263,6 +263,10 @@ func (m *MockHostServices) HTTPFetch(ctx context.Context, method, url string, he
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *MockHostServices) AnyoneFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (m *MockHostServices) GetEnv(ctx context.Context, key string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
@ -575,6 +575,15 @@ type HostServices interface {
|
||||
// HTTP operations
|
||||
HTTPFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error)
|
||||
|
||||
// AnyoneFetch is HTTPFetch routed through the Anyone (ANyONe
|
||||
// protocol) SOCKS5 proxy so the external endpoint sees an Anyone
|
||||
// exit IP, not the gateway's. Feat-11 — server-side analog of the
|
||||
// client-side proxy, for serverless functions fronting third-party
|
||||
// APIs (e.g. wallet RPC) that shouldn't expose a gateway↔upstream
|
||||
// metadata trail. NO silent fallback to direct: returns a typed
|
||||
// error envelope when Anyone routing is unavailable.
|
||||
AnyoneFetch(ctx context.Context, method, url string, headers map[string]string, body []byte) ([]byte, error)
|
||||
|
||||
// Context operations
|
||||
GetEnv(ctx context.Context, key string) (string, error)
|
||||
GetSecret(ctx context.Context, name string) (string, error)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user