diff --git a/agent/store/types/sandbox_v2.go b/agent/store/types/sandbox_v2.go index 9968d613..1a888144 100644 --- a/agent/store/types/sandbox_v2.go +++ b/agent/store/types/sandbox_v2.go @@ -7,9 +7,9 @@ import ( "os" "path/filepath" "sort" - "strings" jsoniter "github.com/json-iterator/go" + "github.com/yaoapp/gou/application" sandboxTypes "github.com/yaoapp/yao/agent/sandbox/v2/types" ) @@ -21,18 +21,9 @@ func LoadSandboxConfig(filePath string) (*sandboxTypes.SandboxConfig, error) { return nil, fmt.Errorf("read sandbox config %s: %w", filePath, err) } - ext := strings.ToLower(filepath.Ext(filePath)) var cfg sandboxTypes.SandboxConfig - - switch ext { - case ".json", ".yao": - if err := jsoniter.Unmarshal(data, &cfg); err != nil { - return nil, fmt.Errorf("parse sandbox config (json): %w", err) - } - default: - if err := jsoniter.Unmarshal(data, &cfg); err != nil { - return nil, fmt.Errorf("parse sandbox config: %w", err) - } + if err := application.Parse(filepath.Base(filePath), data, &cfg); err != nil { + return nil, fmt.Errorf("parse sandbox config: %w", err) } if cfg.Version != sandboxTypes.SandboxVersionV2 { diff --git a/config/types.go b/config/types.go index d7f3bbc5..c339c081 100644 --- a/config/types.go +++ b/config/types.go @@ -14,31 +14,32 @@ func HostHasInternal(host string) bool { // Config 象传应用引擎配置 type Config struct { - Mode string `json:"mode,omitempty" env:"YAO_ENV" envDefault:"production"` // The start mode production/development - AppSource string `json:"app,omitempty" env:"YAO_APP_SOURCE"` // The Application Source Root Path default same as Root - Root string `json:"root,omitempty" env:"YAO_ROOT" envDefault:"."` // The Application Root Path - Lang string `json:"lang,omitempty" env:"YAO_LANG" envDefault:"en-us"` // Default language setting - TimeZone string `json:"timezone,omitempty" env:"YAO_TIMEZONE"` // Default TimeZone - DataRoot string `json:"data_root,omitempty" env:"YAO_DATA_ROOT" envDefault:""` // The data root path - ExtensionRoot string `json:"extension_root,omitempty" env:"YAO_EXTENSION_ROOT" envDefault:""` // Plugin, Wasm root PATH, Default is (/plugins /wasms) - Host string `json:"host,omitempty" env:"YAO_HOST" envDefault:"0.0.0.0"` // The server host - Port int `json:"port,omitempty" env:"YAO_PORT" envDefault:"5099"` // The server port - Cert string `json:"cert,omitempty" env:"YAO_CERT"` // The HTTPS certificate path - Key string `json:"key,omitempty" env:"YAO_KEY"` // The HTTPS certificate key path - Log string `json:"log,omitempty" env:"YAO_LOG"` // The log file path - LogMode string `json:"log_mode,omitempty" env:"YAO_LOG_MODE" envDefault:"TEXT"` // The log mode TEXT|JSON - LogMaxSize int `json:"log_max_size,omitempty" env:"YAO_LOG_MAX_SIZE" envDefault:"100"` // The max log size in MB, the default is 100 - LogMaxAage int `json:"log_max_age,omitempty" env:"YAO_LOG_MAX_AGE" envDefault:"7"` // The max log age in day, the default is 7 - LogMaxBackups int `json:"log_max_backups" env:"YAO_LOG_MAX_BACKUPS" envDefault:"3"` // The max log backups, the default is 3 - LogLocalTime bool `json:"log_local_time" env:"YAO_LOG_LOCAL_TIME" envDefault:"true"` - JWTSecret string `json:"jwt_secret,omitempty" env:"YAO_JWT_SECRET"` // The JWT Secret - DB Database `json:"db,omitempty"` // The database config - AllowFrom []string `json:"allowfrom,omitempty" envSeparator:"|" env:"YAO_ALLOW_FROM"` // Domain list the separator is | - Session Session `json:"session,omitempty"` // Session Config - Runtime Runtime `json:"runtime,omitempty"` // Runtime config - Trace Trace `json:"trace,omitempty"` // Trace config - Registry string `json:"registry,omitempty" env:"YAO_REGISTRY" envDefault:"https://registry.yaoagents.com"` // The package registry server URL - GRPC GRPCConfig `json:"grpc,omitempty"` + Mode string `json:"mode,omitempty" env:"YAO_ENV" envDefault:"production"` // The start mode production/development + AppSource string `json:"app,omitempty" env:"YAO_APP_SOURCE"` // The Application Source Root Path default same as Root + Root string `json:"root,omitempty" env:"YAO_ROOT" envDefault:"."` // The Application Root Path + Lang string `json:"lang,omitempty" env:"YAO_LANG" envDefault:"en-us"` // Default language setting + TimeZone string `json:"timezone,omitempty" env:"YAO_TIMEZONE"` // Default TimeZone + DataRoot string `json:"data_root,omitempty" env:"YAO_DATA_ROOT" envDefault:""` // The data root path + ExtensionRoot string `json:"extension_root,omitempty" env:"YAO_EXTENSION_ROOT" envDefault:""` // Plugin, Wasm root PATH, Default is (/plugins /wasms) + Host string `json:"host,omitempty" env:"YAO_HOST" envDefault:"0.0.0.0"` // The server host + Port int `json:"port,omitempty" env:"YAO_PORT" envDefault:"5099"` // The server port + Cert string `json:"cert,omitempty" env:"YAO_CERT"` // The HTTPS certificate path + Key string `json:"key,omitempty" env:"YAO_KEY"` // The HTTPS certificate key path + Log string `json:"log,omitempty" env:"YAO_LOG"` // The log file path + LogMode string `json:"log_mode,omitempty" env:"YAO_LOG_MODE" envDefault:"TEXT"` // The log mode TEXT|JSON + LogMaxSize int `json:"log_max_size,omitempty" env:"YAO_LOG_MAX_SIZE" envDefault:"100"` // The max log size in MB, the default is 100 + LogMaxAage int `json:"log_max_age,omitempty" env:"YAO_LOG_MAX_AGE" envDefault:"7"` // The max log age in day, the default is 7 + LogMaxBackups int `json:"log_max_backups" env:"YAO_LOG_MAX_BACKUPS" envDefault:"3"` // The max log backups, the default is 3 + LogLocalTime bool `json:"log_local_time" env:"YAO_LOG_LOCAL_TIME" envDefault:"true"` + JWTSecret string `json:"jwt_secret,omitempty" env:"YAO_JWT_SECRET"` // The JWT Secret + DB Database `json:"db,omitempty"` // The database config + AllowFrom []string `json:"allowfrom,omitempty" envSeparator:"|" env:"YAO_ALLOW_FROM"` // Domain list the separator is | + Session Session `json:"session,omitempty"` // Session Config + Runtime Runtime `json:"runtime,omitempty"` // Runtime config + Trace Trace `json:"trace,omitempty"` // Trace config + Registry string `json:"registry,omitempty" env:"YAO_REGISTRY" envDefault:"https://registry.yaoagents.com"` // The package registry server URL + GRPC GRPCConfig `json:"grpc,omitempty"` + HostExec HostExecConfig `json:"host_exec,omitempty"` } // GRPCConfig gRPC server configuration @@ -88,6 +89,15 @@ type Runtime struct { Import bool `json:"import,omitempty" env:"YAO_RUNTIME_IMPORT" envDefault:"true"` // If false the import statement will be disabled, the default value is true. } +// HostExecConfig controls local host execution capability. +type HostExecConfig struct { + Enabled bool `json:"enabled,omitempty" env:"YAO_HOST_EXEC" envDefault:"false"` // Enable host execution on local node + FullAccess bool `json:"full_access,omitempty" env:"YAO_HOST_EXEC_FULL_ACCESS" envDefault:"false"` // Bypass command/dir checks + AllowedCommands []string `json:"allowed_commands,omitempty" env:"YAO_HOST_EXEC_ALLOWED_COMMANDS" envSeparator:","` // Allowed commands (comma-separated) + AllowedDirs []string `json:"allowed_dirs,omitempty" env:"YAO_HOST_EXEC_ALLOWED_DIRS" envSeparator:","` // Allowed working directories + DeniedDirs []string `json:"denied_dirs,omitempty" env:"YAO_HOST_EXEC_DENIED_DIRS" envSeparator:","` // Denied directories (higher priority) +} + // Trace config type Trace struct { Driver string `json:"driver,omitempty" env:"YAO_TRACE_DRIVER"` // The trace driver. local (development) | store (production) diff --git a/engine/load.go b/engine/load.go index 646887dd..d96baf21 100644 --- a/engine/load.go +++ b/engine/load.go @@ -6,6 +6,7 @@ import ( "log" "os" "os/exec" + "path/filepath" "regexp" "strings" "time" @@ -32,7 +33,6 @@ import ( "github.com/yaoapp/yao/kb" "github.com/yaoapp/yao/mcp" "github.com/yaoapp/yao/messenger" - "github.com/yaoapp/yao/moapi" "github.com/yaoapp/yao/model" "github.com/yaoapp/yao/monitor" "github.com/yaoapp/yao/openapi" @@ -45,12 +45,10 @@ import ( "github.com/yaoapp/yao/schedule" "github.com/yaoapp/yao/script" "github.com/yaoapp/yao/share" - "github.com/yaoapp/yao/socket" "github.com/yaoapp/yao/store" sui "github.com/yaoapp/yao/sui/api" - tairegistry "github.com/yaoapp/yao/tai/registry" + "github.com/yaoapp/yao/tai" "github.com/yaoapp/yao/task" - "github.com/yaoapp/yao/websocket" "github.com/yaoapp/yao/widget" "github.com/yaoapp/yao/widgets" @@ -133,20 +131,22 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string warnings = append(warnings, Warning{Widget: "DB", Error: err}) } - // Initialize the Tai node registry (idempotent, safe to call early). - loadStep("Registry", func() error { - tairegistry.InitWithWriter(config.LogOutput, cfg.LogMode) - return nil - }, callback) - - // Initialize the Sandbox manager and start it (auto-registers local Docker - // node if available, recovers existing containers, starts cleanup loop). - err = loadStep("Sandbox", func() error { + // Initialize the Tai registry, register local host node, then start the + // Sandbox manager (container recovery + cleanup loop). + err = loadStep("Registry", func() error { + dataDir := filepath.Join(cfg.DataRoot, "workspaces") + caps := tai.InitLocal(config.LogOutput, cfg.LogMode, dataDir) + if !caps.Docker { + log.Println("[Registry] Docker not available") + } + if caps.HostExec { + log.Println("[Registry] Host execution enabled (YAO_HOST_EXEC=true)") + } sandbox.Init() return sandbox.M().Start(context.Background()) }, callback) if err != nil { - warnings = append(warnings, Warning{Widget: "Sandbox", Error: err}) + warnings = append(warnings, Warning{Widget: "Registry", Error: err}) } // Load Certs @@ -300,22 +300,6 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string warnings = append(warnings, Warning{Widget: "API", Error: err}) } - // Load Sockets - err = loadStep("Socket", func() error { - return socket.Load(cfg) - }, callback) - if err != nil { - warnings = append(warnings, Warning{Widget: "Socket", Error: err}) - } - - // Load websockets (client mode) - err = loadStep("WebSocket", func() error { - return websocket.Load(cfg) - }, callback) - if err != nil { - warnings = append(warnings, Warning{Widget: "WebSocket", Error: err}) - } - // Load tasks err = loadStep("Task", func() error { return task.Load(cfg) @@ -364,14 +348,6 @@ func Load(cfg config.Config, options LoadOption, progressCallback ...func(string warnings = append(warnings, Warning{Widget: "SUI", Error: err}) } - // Load Moapi - err = loadStep("Moapi", func() error { - return moapi.Load(cfg) - }, callback) - if err != nil { - warnings = append(warnings, Warning{Widget: "Moapi", Error: err}) - } - // Load Pipe err = loadStep("Pipe", func() error { return pipe.Load(cfg) @@ -496,8 +472,6 @@ func Unload() (err error) { // importers // tasks // schedules - // sockets - // websockets // widgets // custom widget @@ -605,18 +579,6 @@ func Reload(cfg config.Config, options LoadOption) (err error) { printErr(cfg.Mode, "API", err) } - // Load Sockets - err = socket.Load(cfg) // Load sockets - if err != nil { - printErr(cfg.Mode, "Socket", err) - } - - // Load websockets (client mode) - err = websocket.Load(cfg) - if err != nil { - printErr(cfg.Mode, "WebSocket", err) - } - // Load tasks err = task.Load(cfg) if err != nil { diff --git a/moapi/api.go b/moapi/api.go deleted file mode 100644 index 92b86647..00000000 --- a/moapi/api.go +++ /dev/null @@ -1,36 +0,0 @@ -package moapi - -import "github.com/yaoapp/gou/api" - -var dsl = []byte(` -{ - "name": "Moapi API", - "description": "The API for Moapi", - "version": "1.0.0", - "guard": "bearer-jwt", - "group": "__moapi/v1", - "paths": [ - { - "path": "/images/generations", - "method": "POST", - "process": "moapi.images.Generations", - "in": ["$payload.model", "$payload.prompt", ":payload"], - "out": { "status": 200, "type": "application/json" } - }, - - { - "path": "/chat/completions", - "guard": "query-jwt", - "method": "GET", - "process": "moapi.chat.Completions", - "processHandler": true, - "out": { "status": 200, "type": "text/event-stream" } - } - ] -} -`) - -func registerAPI() error { - _, err := api.LoadSource(".yao", dsl, "moapi.v1") - return err -} diff --git a/moapi/moapi.go b/moapi/moapi.go deleted file mode 100644 index d94fd916..00000000 --- a/moapi/moapi.go +++ /dev/null @@ -1,169 +0,0 @@ -package moapi - -// *** WARNING *** -// Temporarily: change after the moapi is open source -// - -import ( - "fmt" - - jsoniter "github.com/json-iterator/go" - "github.com/yaoapp/gou/http" - "github.com/yaoapp/yao/config" - "github.com/yaoapp/yao/share" -) - -// Mirrors list all the mirrors -var cacheMirrors = []*Mirror{} -var cacheApps = []*App{} -var cacheMirrorsMap = map[string]*Mirror{} - -// Models list all the models -var Models = []string{ - "gpt-4-1106-preview", - "gpt-4-1106-vision-preview", - "gpt-4", - "gpt-4-32k", - - "gpt-3.5-turbo", - "gpt-3.5-turbo-1106", - "gpt-3.5-turbo-instruct", - - "dall-e-3", - "dall-e-2", - - "tts-1", - "tts-1-hd", - - "text-moderation-latest", - "text-moderation-stable", - - "text-embedding-ada-002", - "whisper-1", -} - -// Load load the moapi -func Load(cfg config.Config) error { - return registerAPI() -} - -// Mirrors list all the mirrors -func Mirrors(cache bool) ([]*Mirror, error) { - if cache && len(cacheMirrors) > 0 { - return cacheMirrors, nil - } - - bytes, err := httpGet("/api/moapi/mirrors") - if err != nil { - return nil, err - } - - err = jsoniter.Unmarshal(bytes, &cacheMirrors) - if err != nil { - return nil, err - } - - for _, mirror := range cacheMirrors { - cacheMirrorsMap[mirror.Host] = mirror - } - - return cacheMirrors, nil -} - -// Apps list all the apps -func Apps(cache bool) ([]*App, error) { - if cache && len(cacheApps) > 0 { - return cacheApps, nil - } - - mirrors := SelectMirrors() - bytes, err := httpGet("/api/moapi/apps", mirrors...) - if err != nil { - return nil, err - } - - err = jsoniter.Unmarshal(bytes, &cacheApps) - if err != nil { - return nil, err - } - - channel := Channel() - if channel != "" { - for i := range cacheApps { - cacheApps[i].Homepage = cacheApps[i].Homepage + "?channel=" + channel - } - } - - return cacheApps, nil -} - -// Homepage get the home page url with the invite code -func Homepage() string { - channel := Channel() - if channel == "" { - return "https://store.moapi.ai" - } - return "https://store.moapi.ai" + "?channel=" + channel -} - -// Channel get the channel -func Channel() string { - - return share.App.Moapi.Channel -} - -// SelectMirrors select the mirrors -func SelectMirrors() []*Mirror { - - if share.App.Moapi.Mirrors == nil || len(share.App.Moapi.Mirrors) == 0 { - return []*Mirror{} - } - - _, err := Mirrors(true) - if err != nil { - return []*Mirror{} - } - - // pick the mirrors - var result []*Mirror - for _, host := range share.App.Moapi.Mirrors { - if mirror, ok := cacheMirrorsMap[host]; ok { - if mirror.Status == "on" { - result = append(result, mirror) - } - } - } - - return result -} - -// httpGet get the data from the api -func httpGet(api string, mirrors ...*Mirror) ([]byte, error) { - return httpGetRetry(api, mirrors, 0) -} - -func httpGetRetry(api string, mirrors []*Mirror, retryTimes int) ([]byte, error) { - - url := "https://" + share.MoapiHosts[retryTimes] + api - if len(mirrors) > retryTimes { - url = "https://" + mirrors[retryTimes].Host + api - } - - secret := share.App.Moapi.Secret - organization := share.App.Moapi.Organization - - http := http.New(url) - http.SetHeader("Authorization", "Bearer "+secret) - http.SetHeader("Content-Type", "application/json") - http.SetHeader("Moapi-Organization", organization) - - resp := http.Get() - if resp.Code >= 500 { - if retryTimes > 3 { - return nil, fmt.Errorf("Moapi Server Error: %s", resp.Data) - } - return httpGetRetry(api, mirrors, retryTimes+1) - } - - return jsoniter.Marshal(resp.Data) -} diff --git a/moapi/process.go b/moapi/process.go deleted file mode 100644 index e6ba8127..00000000 --- a/moapi/process.go +++ /dev/null @@ -1,146 +0,0 @@ -package moapi - -import ( - "context" - "io" - "net/http" - "strings" - - "github.com/gin-gonic/gin" - jsoniter "github.com/json-iterator/go" - "github.com/yaoapp/gou/process" - "github.com/yaoapp/kun/exception" - "github.com/yaoapp/yao/openai" -) - -func init() { - process.RegisterGroup("moapi", map[string]process.Handler{ - "images.generations": ImagesGenerations, - "chat.completions": ChatCompletions, - }) -} - -// ImagesGenerations Generate images -func ImagesGenerations(process *process.Process) interface{} { - - process.ValidateArgNums(2) - model := process.ArgsString(0) - prompt := process.ArgsString(1) - option := process.ArgsMap(2, map[string]interface{}{}) - - if model == "" { - exception.New("ImagesGenerations error: model is required", 400).Throw() - } - - if prompt == "" { - exception.New("ImagesGenerations error: prompt is required", 400).Throw() - } - - ai, err := openai.NewMoapi(model) - if err != nil { - exception.New("ImagesGenerations error: %s", 400, err).Throw() - } - - option["model"] = model - res, ex := ai.ImagesGenerations(prompt, option) - if ex != nil { - ex.Throw() - } - return res -} - -// ChatCompletions chat completions -func ChatCompletions(process *process.Process) interface{} { - - return func(c *gin.Context) { - - option := map[string]interface{}{} - query := c.Query("payload") - err := jsoniter.UnmarshalFromString(query, &option) - if err != nil { - exception.New("ChatCompletions error: %s", 400, err).Throw() - } - - // option := payload - // model := "gpt-3.5-turbo" - // messages := []map[string]interface{}{ - // { - // "role": "system", - // "content": "You are a helpful assistant.", - // }, - // { - // "role": "user", - // "content": "Hello!", - // }, - // // } - - // option["messages"] = messages - // option["model"] = model - - delete(option, "context") - model, ok := option["model"].(string) - if !ok || model == "" { - exception.New("ChatCompletions error: model is required", 400).Throw() - } - - ai, err := openai.NewMoapi(model) - if err != nil { - exception.New("ChatCompletions error: %s", 400, err).Throw() - } - - if v, ok := option["stream"].(bool); ok && v { - - chanStream := make(chan []byte, 1) - chanError := make(chan error, 1) - - defer func() { - close(chanStream) - close(chanError) - }() - - ctx, cancel := context.WithCancel(c.Request.Context()) - defer cancel() - - go ai.Stream(ctx, "/v1/chat/completions", option, func(data []byte) int { - - if (string(data)) == "\n" || string(data) == "" { - return 1 // HandlerReturnOk - } - - chanStream <- data - if strings.HasSuffix(string(data), "[DONE]") { - return 0 // HandlerReturnBreak0 - } - return 1 // HandlerReturnOk - }) - - c.Header("Content-Type", "text/event-stream") - c.Stream(func(w io.Writer) bool { - select { - case err := <-chanError: - if err != nil { - c.JSON(http.StatusInternalServerError, err.Error()) - } - return false - - case msg := <-chanStream: - - if string(msg) == "\n" { - return true - } - - message := strings.TrimLeft(string(msg), "data: ") - c.SSEvent("message", message) - return true - - case <-ctx.Done(): - return false - } - }) - - return - } - - return - } -} diff --git a/moapi/types.go b/moapi/types.go deleted file mode 100644 index 8d0459f8..00000000 --- a/moapi/types.go +++ /dev/null @@ -1,34 +0,0 @@ -package moapi - -// Mirror is the mirror info -type Mirror struct { - Name string `json:"name"` - Host string `json:"host"` - Area string `json:"area"` // area code - Latency int `json:"latency"` // ms - Status string `json:"status"` // on, slow, off, -} - -// App is the app info -type App struct { - Name string `json:"name"` - UpdatedAt int64 `json:"updated_at"` - CreatedAt int64 `json:"created_at"` - Country string `json:"country"` - Creator string `json:"creator"` - Description string `json:"description"` - Version string `json:"version"` - Short string `json:"short"` - Icon string `json:"icon"` - Homepage string `json:"homepage"` - Images []string `json:"images,omitempty"` - Videos []string `json:"videos,omitempty"` - Stat AppStat `json:"stat,omitempty"` - Languages []string `json:"languages"` -} - -// AppStat is the app stat info -type AppStat struct { - Downloads int `json:"downloads"` - Stars int `json:"stars"` -} diff --git a/openapi/computer/computer.go b/openapi/computer/computer.go index 09311000..4ce6d4a5 100644 --- a/openapi/computer/computer.go +++ b/openapi/computer/computer.go @@ -89,7 +89,7 @@ func handleOptions(c *gin.Context) { if kindFilter == "" || kindFilter == "host" { for i := range snaps { s := &snaps[i] - if !nodeOwnedBy(s, authInfo) { + if s.Mode != "local" && !nodeOwnedBy(s, authInfo) { continue } if !s.Capabilities.HostExec { @@ -106,7 +106,7 @@ func handleOptions(c *gin.Context) { if kindFilter == "" || kindFilter == "node" { for i := range snaps { s := &snaps[i] - if !nodeOwnedBy(s, authInfo) { + if s.Mode != "local" && !nodeOwnedBy(s, authInfo) { continue } hasRuntime := s.Capabilities.Docker || s.Capabilities.K8s diff --git a/openapi/nodes/nodes.go b/openapi/nodes/nodes.go index 1161a002..f662f31e 100644 --- a/openapi/nodes/nodes.go +++ b/openapi/nodes/nodes.go @@ -98,19 +98,15 @@ func handleList(c *gin.Context) { } authInfo := authorized.GetInfo(c) - - var snaps []taitypes.NodeMeta - if authInfo != nil && authInfo.TeamID != "" { - snaps = reg.ListByTeam(authInfo.TeamID) - } else if authInfo != nil && authInfo.UserID != "" { - snaps = reg.ListByUser(authInfo.UserID) - } else { - snaps = reg.List() - } + snaps := reg.List() result := make([]nodeResponse, 0, len(snaps)) - for _, s := range snaps { - result = append(result, snapToResponse(s)) + for i := range snaps { + s := &snaps[i] + if s.Mode != "local" && !nodeOwnedBy(s, authInfo) { + continue + } + result = append(result, snapToResponse(*s)) } response.RespondWithSuccess(c, http.StatusOK, result) } @@ -130,7 +126,7 @@ func handleGet(c *gin.Context) { } authInfo := authorized.GetInfo(c) - if !nodeOwnedBy(snap, authInfo) { + if snap.Mode != "local" && !nodeOwnedBy(snap, authInfo) { c.JSON(http.StatusForbidden, gin.H{"error": "no permission to access this node"}) return } diff --git a/openapi/sandbox/manage.go b/openapi/sandbox/manage.go index 3df56482..0235f37d 100644 --- a/openapi/sandbox/manage.go +++ b/openapi/sandbox/manage.go @@ -251,15 +251,15 @@ func handleList(c *gin.Context) { var result []sandboxResponse - // Host entries: list all nodes, filter by ownership + host_exec + // Host entries: list registered nodes that have any compute capability. if reg := registry.Global(); reg != nil { snaps := reg.List() for i := range snaps { s := &snaps[i] - if !nodeOwnedBy(s, authInfo) { + if s.Mode != "local" && !nodeOwnedBy(s, authInfo) { continue } - if !s.Capabilities.HostExec { + if !s.Capabilities.HostExec && !s.Capabilities.Docker { continue } if nodeFilter != "" && s.TaiID != nodeFilter { diff --git a/sandbox/v2/manager.go b/sandbox/v2/manager.go index b25f6976..1e58bd07 100644 --- a/sandbox/v2/manager.go +++ b/sandbox/v2/manager.go @@ -4,13 +4,11 @@ import ( "context" "fmt" "log" - "path/filepath" goruntime "runtime" "strconv" "sync" "time" - "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/tai" "github.com/yaoapp/yao/tai/registry" tairuntime "github.com/yaoapp/yao/tai/runtime" @@ -29,16 +27,14 @@ func newManager() *Manager { } // Start discovers existing containers from all registered nodes and rebuilds -// the boxes map. If no "local" node is registered yet, it probes the local -// Docker environment and auto-registers one when available. +// the boxes map. The local node must already be registered by tai.InitLocal() +// before Start is called. func (m *Manager) Start(ctx context.Context) error { reg := registry.Global() if reg == nil { return nil } - m.ensureLocalNode(reg) - for _, snap := range reg.List() { res, err := m.getNode(snap.TaiID) if err != nil { @@ -50,15 +46,6 @@ func (m *Manager) Start(ctx context.Context) error { return nil } -// ensureLocalNode delegates to tai.RegisterLocal() which probes the local -// Docker environment and registers a "local" node in the registry if available. -// The workspace data directory is derived from config.Conf.DataRoot so that -// workspace files persist across restarts. -func (m *Manager) ensureLocalNode(_ *registry.Registry) { - dataDir := filepath.Join(config.Conf.DataRoot, "workspaces") - tai.RegisterLocal(tai.WithDataDir(dataDir)) -} - // Nodes returns the list of registered Tai nodes from the registry. func (m *Manager) Nodes() []taitypes.NodeMeta { reg := registry.Global() @@ -148,6 +135,11 @@ func (m *Manager) Create(ctx context.Context, opts CreateOptions) (*Box, error) nodeID = targetNode } else { nodeID = node + if nodeID == "local" { + if ws, e := wsm.Get(ctx, opts.WorkspaceID); e == nil && ws.Owner != "" && ws.Owner != opts.Owner { + return nil, fmt.Errorf("sandbox: no permission to mount workspace %q", opts.WorkspaceID) + } + } } } } diff --git a/socket/socket.go b/socket/socket.go deleted file mode 100644 index 4a9ab9f4..00000000 --- a/socket/socket.go +++ /dev/null @@ -1,31 +0,0 @@ -package socket - -import ( - "github.com/yaoapp/yao/config" -) - -// Load 加载API -func Load(cfg config.Config) error { - // var root = filepath.Join(cfg.Root, "sockets") - // return LoadFrom(root, "") - return nil -} - -// LoadFrom 从特定目录加载 -// func LoadFrom(dir string, prefix string) error { - -// if share.DirNotExists(dir) { -// return fmt.Errorf("%s does not exists", dir) -// } - -// err := share.Walk(dir, ".sock.json", func(root, filename string) { -// name := prefix + share.SpecName(root, filename) -// content := share.ReadFile(filename) -// _, err := gou.LoadSocket(string(content), name) -// if err != nil { -// log.With(log.F{"root": root, "file": filename}).Error(err.Error()) -// } -// }) - -// return err -// } diff --git a/socket/socket_test.go b/socket/socket_test.go deleted file mode 100644 index e7c142d1..00000000 --- a/socket/socket_test.go +++ /dev/null @@ -1,22 +0,0 @@ -package socket - -import ( - "testing" - - "github.com/stretchr/testify/assert" - "github.com/yaoapp/gou/socket" - "github.com/yaoapp/yao/config" -) - -func TestLoad(t *testing.T) { - Load(config.Conf) - check(t) -} - -func check(t *testing.T) { - keys := []string{} - for key := range socket.Sockets { - keys = append(keys, key) - } - assert.Equal(t, 0, len(keys)) -} diff --git a/tai/dial.go b/tai/dial.go index 0f560cd3..7937d801 100644 --- a/tai/dial.go +++ b/tai/dial.go @@ -7,6 +7,8 @@ import ( "net/http" "time" + yaoconfig "github.com/yaoapp/yao/config" + "github.com/yaoapp/yao/tai/hostexec" hepb "github.com/yaoapp/yao/tai/hostexec/pb" "github.com/yaoapp/yao/tai/proxy" "github.com/yaoapp/yao/tai/registry" @@ -88,15 +90,17 @@ func DialTunnel(taiID string, reg *registry.Registry, opts ...DialOption) (*Conn return res, nil } -// DialLocal establishes connections to the local Docker daemon. +// DialLocal establishes connections to the local host as a Tai node. +// Docker is probed but not required — when unavailable the node still +// provides Volume (and optionally HostExec) capabilities. // Does NOT interact with the registry. Caller must call ConnResources.Close(). func DialLocal(addr string, dataDir string, vol volume.Volume) (*ConnResources, error) { - sb, err := runtime.NewLocal(addr) - if err != nil && vol == nil { - return nil, err - } + sb, _ := runtime.NewLocal(addr) // Docker failure is non-fatal - res := &ConnResources{DataDir: dataDir} + res := &ConnResources{ + DataDir: dataDir, + System: CollectSystemInfo(), + } if sb != nil { res.Runtime = sb @@ -105,6 +109,15 @@ func DialLocal(addr string, dataDir string, vol volume.Volume) (*ConnResources, res.VNC = vnc.NewLocal(sb) } + if yaoconfig.Conf.HostExec.Enabled { + res.HostExec = hostexec.NewLocalClient(dataDir, hostexec.Policy{ + FullAccess: yaoconfig.Conf.HostExec.FullAccess, + AllowedCommands: yaoconfig.Conf.HostExec.AllowedCommands, + AllowedDirs: yaoconfig.Conf.HostExec.AllowedDirs, + DeniedDirs: yaoconfig.Conf.HostExec.DeniedDirs, + }) + } + if vol != nil { res.Volume = vol } else { diff --git a/tai/hostexec/local.go b/tai/hostexec/local.go new file mode 100644 index 00000000..32174344 --- /dev/null +++ b/tai/hostexec/local.go @@ -0,0 +1,324 @@ +package hostexec + +import ( + "bytes" + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + pb "github.com/yaoapp/yao/tai/hostexec/pb" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +const defaultMaxOutputBytes = 10 * 1024 * 1024 // 10 MB + +// Policy controls which commands and directories are allowed. +type Policy struct { + FullAccess bool // bypass command and path checks + AllowedCommands []string // empty = all denied (unless FullAccess) + AllowedDirs []string // working_dir must be under one of these + DeniedDirs []string // higher priority than AllowedDirs +} + +// --------------------------------------------------------------------------- +// LocalClient — in-process HostExecClient (no gRPC network hop) +// --------------------------------------------------------------------------- + +// LocalClient implements pb.HostExecClient by executing commands directly on +// the current host via os/exec. +type LocalClient struct { + defaultDir string + policy Policy +} + +// Compile-time interface check. +var _ pb.HostExecClient = (*LocalClient)(nil) + +// NewLocalClient creates a LocalClient. +func NewLocalClient(defaultDir string, policy Policy) *LocalClient { + return &LocalClient{defaultDir: defaultDir, policy: policy} +} + +// Exec runs a command synchronously and returns the result. +func (c *LocalClient) Exec(ctx context.Context, req *pb.ExecRequest, _ ...grpc.CallOption) (*pb.ExecResponse, error) { + if err := c.checkCommand(req.Command); err != nil { + return &pb.ExecResponse{Error: err.Error()}, nil + } + if err := c.checkWorkingDir(req.WorkingDir); err != nil { + return &pb.ExecResponse{Error: err.Error()}, nil + } + + timeout := time.Duration(req.TimeoutMs) * time.Millisecond + if timeout <= 0 { + timeout = 5 * time.Minute + } + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + cmd := exec.CommandContext(ctx, req.Command, req.Args...) + cmd.Dir = c.resolveDir(req.WorkingDir) + cmd.Env = c.buildEnv(req.Env) + if len(req.Stdin) > 0 { + cmd.Stdin = bytes.NewReader(req.Stdin) + } + + maxBytes := req.MaxOutputBytes + if maxBytes <= 0 { + maxBytes = defaultMaxOutputBytes + } + + var stdout, stderr bytes.Buffer + cmd.Stdout = &limitWriter{buf: &stdout, max: maxBytes} + cmd.Stderr = &limitWriter{buf: &stderr, max: maxBytes} + + start := time.Now() + err := cmd.Run() + + resp := &pb.ExecResponse{ + Stdout: stdout.Bytes(), + Stderr: stderr.Bytes(), + DurationMs: time.Since(start).Milliseconds(), + } + if int64(len(resp.Stdout)+len(resp.Stderr)) >= maxBytes { + resp.Truncated = true + } + + if err != nil { + if ctx.Err() != nil { + resp.Error = "command timed out" + resp.ExitCode = -1 + } else if exitErr, ok := err.(*exec.ExitError); ok { + resp.ExitCode = int32(exitErr.ExitCode()) + } else { + resp.Error = err.Error() + resp.ExitCode = -1 + } + } + return resp, nil +} + +// ExecStream runs a command and streams stdout/stderr via a channel-based +// adapter that satisfies grpc.ServerStreamingClient[pb.ExecOutput]. +func (c *LocalClient) ExecStream(ctx context.Context, req *pb.ExecRequest, _ ...grpc.CallOption) (grpc.ServerStreamingClient[pb.ExecOutput], error) { + if err := c.checkCommand(req.Command); err != nil { + return newErrorStream(ctx, err.Error()), nil + } + if err := c.checkWorkingDir(req.WorkingDir); err != nil { + return newErrorStream(ctx, err.Error()), nil + } + + timeout := time.Duration(req.TimeoutMs) * time.Millisecond + if timeout <= 0 { + timeout = 5 * time.Minute + } + ctx, cancel := context.WithTimeout(ctx, timeout) + + cmd := exec.CommandContext(ctx, req.Command, req.Args...) + cmd.Dir = c.resolveDir(req.WorkingDir) + cmd.Env = c.buildEnv(req.Env) + if len(req.Stdin) > 0 { + cmd.Stdin = bytes.NewReader(req.Stdin) + } + + stdoutPipe, err := cmd.StdoutPipe() + if err != nil { + cancel() + return newErrorStream(ctx, err.Error()), nil + } + stderrPipe, err := cmd.StderrPipe() + if err != nil { + cancel() + return newErrorStream(ctx, err.Error()), nil + } + + if err := cmd.Start(); err != nil { + cancel() + return newErrorStream(ctx, err.Error()), nil + } + + ch := make(chan *pb.ExecOutput, 64) + go func() { + defer cancel() + defer close(ch) + + done := make(chan struct{}) + go func() { + defer close(done) + streamPipe(ch, stdoutPipe, pb.ExecOutput_STDOUT) + }() + streamPipe(ch, stderrPipe, pb.ExecOutput_STDERR) + <-done + + waitErr := cmd.Wait() + final := &pb.ExecOutput{Done: true} + if waitErr != nil { + if exitErr, ok := waitErr.(*exec.ExitError); ok { + final.ExitCode = int32(exitErr.ExitCode()) + } else { + final.Error = waitErr.Error() + final.ExitCode = -1 + } + } + ch <- final + }() + + return &localStream{ctx: ctx, ch: ch}, nil +} + +// --------------------------------------------------------------------------- +// Policy checks (identical to Tai hostexec/server.go) +// --------------------------------------------------------------------------- + +func (c *LocalClient) checkCommand(command string) error { + if c.policy.FullAccess { + return nil + } + if len(c.policy.AllowedCommands) == 0 { + return fmt.Errorf("hostexec: no commands are allowed (allowed_commands is empty)") + } + base := filepath.Base(command) + for _, allowed := range c.policy.AllowedCommands { + if command == allowed || base == allowed { + return nil + } + } + return fmt.Errorf("hostexec: command %q is not in the allowed list", command) +} + +func (c *LocalClient) checkWorkingDir(dir string) error { + if dir == "" || c.policy.FullAccess { + return nil + } + absDir, err := filepath.Abs(dir) + if err != nil { + return fmt.Errorf("hostexec: invalid working_dir %q: %w", dir, err) + } + resolved, err := filepath.EvalSymlinks(absDir) + if err != nil { + resolved = absDir + } + for _, denied := range c.policy.DeniedDirs { + if matchDir(resolved, denied) { + return fmt.Errorf("hostexec: working_dir %q is in a denied directory", dir) + } + } + if len(c.policy.AllowedDirs) == 0 { + return nil + } + for _, allowed := range c.policy.AllowedDirs { + if matchDir(resolved, allowed) { + return nil + } + } + return fmt.Errorf("hostexec: working_dir %q is not in any allowed directory", dir) +} + +func matchDir(resolved, dir string) bool { + absDir, _ := filepath.Abs(dir) + resolvedDir, err := filepath.EvalSymlinks(absDir) + if err != nil { + resolvedDir = absDir + } + if resolved == resolvedDir { + return true + } + return strings.HasPrefix(resolved, resolvedDir+string(filepath.Separator)) +} + +func (c *LocalClient) resolveDir(dir string) string { + if dir != "" { + return dir + } + if c.defaultDir != "" { + return c.defaultDir + } + return "" +} + +func (c *LocalClient) buildEnv(userEnv map[string]string) []string { + env := os.Environ() + for k, v := range userEnv { + env = append(env, k+"="+v) + } + return env +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +func streamPipe(ch chan<- *pb.ExecOutput, pipe io.ReadCloser, st pb.ExecOutput_Stream) { + buf := make([]byte, 32*1024) + for { + n, err := pipe.Read(buf) + if n > 0 { + data := make([]byte, n) + copy(data, buf[:n]) + ch <- &pb.ExecOutput{Stream: st, Data: data} + } + if err != nil { + return + } + } +} + +type limitWriter struct { + buf *bytes.Buffer + max int64 +} + +func (w *limitWriter) Write(p []byte) (int, error) { + remaining := w.max - int64(w.buf.Len()) + if remaining <= 0 { + return len(p), nil + } + if int64(len(p)) > remaining { + p = p[:remaining] + } + return w.buf.Write(p) +} + +// --------------------------------------------------------------------------- +// localStream — channel-based grpc.ServerStreamingClient adapter +// --------------------------------------------------------------------------- + +type localStream struct { + ctx context.Context + ch <-chan *pb.ExecOutput +} + +var _ grpc.ServerStreamingClient[pb.ExecOutput] = (*localStream)(nil) + +func (s *localStream) Recv() (*pb.ExecOutput, error) { + select { + case <-s.ctx.Done(): + return nil, s.ctx.Err() + case msg, ok := <-s.ch: + if !ok { + return nil, io.EOF + } + return msg, nil + } +} + +func (s *localStream) Header() (metadata.MD, error) { return nil, nil } +func (s *localStream) Trailer() metadata.MD { return nil } +func (s *localStream) CloseSend() error { return nil } +func (s *localStream) Context() context.Context { return s.ctx } +func (s *localStream) SendMsg(any) error { return nil } +func (s *localStream) RecvMsg(any) error { return nil } + +// newErrorStream returns a stream that yields a single Done message with the +// given error, then EOF. Used for early policy-check failures. +func newErrorStream(ctx context.Context, errMsg string) grpc.ServerStreamingClient[pb.ExecOutput] { + ch := make(chan *pb.ExecOutput, 1) + ch <- &pb.ExecOutput{Done: true, Error: errMsg, ExitCode: -1} + close(ch) + return &localStream{ctx: ctx, ch: ch} +} diff --git a/tai/sysinfo.go b/tai/sysinfo.go new file mode 100644 index 00000000..04ad22f0 --- /dev/null +++ b/tai/sysinfo.go @@ -0,0 +1,37 @@ +package tai + +import ( + "os" + "os/exec" + goruntime "runtime" + + "github.com/yaoapp/yao/tai/types" +) + +// CollectSystemInfo gathers system information for the local host. +// The result is identical in structure to what a remote Tai node reports +// via the ServerInfo gRPC service, keeping local and remote nodes symmetric. +func CollectSystemInfo() types.SystemInfo { + hostname, _ := os.Hostname() + return types.SystemInfo{ + OS: goruntime.GOOS, + Arch: goruntime.GOARCH, + Hostname: hostname, + NumCPU: goruntime.NumCPU(), + Shell: detectShell(), + TempDir: os.TempDir(), + } +} + +func detectShell() string { + if goruntime.GOOS != "windows" { + return "sh" + } + if _, err := exec.LookPath("pwsh"); err == nil { + return "pwsh" + } + if _, err := exec.LookPath("powershell"); err == nil { + return "powershell" + } + return "cmd.exe" +} diff --git a/tai/tai.go b/tai/tai.go index 64ccf6d2..ceec75d4 100644 --- a/tai/tai.go +++ b/tai/tai.go @@ -1,6 +1,8 @@ package tai import ( + "io" + "github.com/yaoapp/yao/tai/registry" "github.com/yaoapp/yao/tai/types" "github.com/yaoapp/yao/tai/volume" @@ -69,10 +71,11 @@ func intOr(v, fallback int) int { return fallback } -// RegisterLocal probes the local Docker environment and, if reachable, -// registers it as the "local" node in the registry with ConnResources. -// Returns true if a local node was successfully registered. -// Silently returns false if Docker is not available — this is not an error. +// RegisterLocal probes the local environment and registers the current host +// as the "local" node. Capabilities are set based on actual availability: +// Docker is probed, HostExec is controlled by YAO_HOST_EXEC env var. +// Always returns true — the local node is always registered (at minimum +// with Volume capability). func RegisterLocal(opts ...Option) bool { reg := registry.Global() if reg == nil { @@ -93,13 +96,35 @@ func RegisterLocal(opts ...Option) bool { } reg.Register(®istry.TaiNode{ - TaiID: "local", - Mode: "local", + TaiID: "local", + Mode: "local", + System: res.System, + Capabilities: types.Capabilities{ + Docker: res.Runtime != nil, + HostExec: res.HostExec != nil, + }, }) reg.SetResources("local", res) return true } +// InitLocal initializes the Tai registry and registers the local host as a +// node in a single call. This is the preferred entry point for application +// startup. +// +// Capabilities are determined by probing the environment: +// - Docker reachable → Docker capability +// - YAO_HOST_EXEC=true → HostExec capability (with Policy from env) +// - Volume is always available +func InitLocal(w io.Writer, logMode string, dataDir string) types.Capabilities { + registry.InitWithWriter(w, logMode) + RegisterLocal(WithDataDir(dataDir)) + if meta, ok := registry.Global().Get("local"); ok { + return meta.Capabilities + } + return types.Capabilities{} +} + // GetResources returns the ConnResources for a registered Tai node. func GetResources(taiID string) (*ConnResources, bool) { reg := registry.Global() diff --git a/websocket/websocket.go b/websocket/websocket.go deleted file mode 100644 index 991cf40f..00000000 --- a/websocket/websocket.go +++ /dev/null @@ -1,38 +0,0 @@ -package websocket - -import ( - "github.com/yaoapp/yao/config" -) - -// Load 加载API -func Load(cfg config.Config) error { - // exts := []string{"*.http.yao", "*.http.json", "*.http.jsonc"} - // return application.App.Walk("websockets", func(root, file string, isdir bool) error { - // _, err := websocket.Load(file, share.ID(root, file)) - // return err - // }, exts...) - - // var root = filepath.Join(cfg.Root, "websockets") - // return LoadFrom(root, "") - - return nil -} - -// // LoadFrom 从特定目录加载 -// func LoadFrom(dir string, prefix string) error { - -// if share.DirNotExists(dir) { -// return fmt.Errorf("%s does not exists", dir) -// } - -// err := share.Walk(dir, ".ws.json", func(root, filename string) { -// name := prefix + share.SpecName(root, filename) -// content := share.ReadFile(filename) -// _, err := gou.LoadWebSocket(string(content), name) -// if err != nil { -// log.With(log.F{"root": root, "file": filename}).Error(err.Error()) -// } -// }) - -// return err -// } diff --git a/websocket/websocket_test.go b/websocket/websocket_test.go deleted file mode 100644 index 047674cc..00000000 --- a/websocket/websocket_test.go +++ /dev/null @@ -1,65 +0,0 @@ -package websocket - -import ( - "fmt" - "net" - "net/http" - "testing" - "time" - - "github.com/gin-gonic/gin" - "github.com/yaoapp/gou/websocket" - "github.com/yaoapp/yao/config" -) - -func TestLoad(t *testing.T) { - Load(config.Conf) - check(t) -} - -func TestWebSocketOpen(t *testing.T) { - // Load(config.Conf) - // script.Load(config.Conf) - // srv, url := serve(t) - // defer srv.Stop() - - // ws := websocket.Se("message") - // err := ws.Open(url, "messageV2", "chatV3") - // if err != nil { - // t.Fatal(err) - // } -} - -func serve(t *testing.T) (*websocket.Upgrader, string) { - - ws, err := websocket.NewUpgrader("test") - if err != nil { - t.Fatalf("%s", err) - } - - gin.SetMode(gin.ReleaseMode) - router := gin.Default() - ws.SetHandler(func(message []byte, id int) ([]byte, error) { return message, nil }) - ws.SetRouter(router) - - listener, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - - go ws.Start() - go func() { - http.Serve(listener, router) - }() - time.Sleep(200 * time.Millisecond) - - return ws, fmt.Sprintf("ws://127.0.0.1:%d/websocket/test", listener.Addr().(*net.TCPAddr).Port) -} - -func check(t *testing.T) { - // keys := []string{} - // for key := range gou.WebSockets { - // keys = append(keys, key) - // } - // assert.Equal(t, 1, len(keys)) -}