diff --git a/app.go b/app.go
index 8e3daf21f..ac7a7542e 100755
--- a/app.go
+++ b/app.go
@@ -35,6 +35,7 @@ import (
"os/signal"
"runtime"
"runtime/debug"
+ "strings"
"sync"
"syscall"
"time"
@@ -184,6 +185,14 @@ func (app *App) initWithFlags() {
flag.StringVar(&app.svcFlag, "service", "", "service management, options: install,uninstall,start,stop")
flag.StringVar(&app.svcUser, "service-user", "", "OS user account used to run the service")
+ // -e KEY=VALUE (repeatable): environment overrides applied BEFORE the
+ // config loads. Picked up by $[[env.KEY]] template expansion in the
+ // YAML (OS env wins over the YAML env: section), so deployments can
+ // parameterize any templated setting from the command line instead of
+ // editing config files. Example:
+ // agent -e MANAGED=true -e REMOTE_CONFIG_SERVERS=http://lp:29000
+ flag.Var(&cliEnvOverrides{}, "e", "environment override KEY=VALUE (repeatable, feeds $[[env.KEY]])")
+
if debugFlagInitFunc != nil {
debugFlagInitFunc()
}
@@ -622,3 +631,20 @@ func (app *App) Run() {
log.Error(err)
}
}
+
+// cliEnvOverrides collects repeatable -e KEY=VALUE flags; Apply sets them
+// into the process environment before configuration loading begins.
+type cliEnvOverrides struct {
+ pairs []string
+}
+
+func (c *cliEnvOverrides) String() string { return strings.Join(c.pairs, " ") }
+
+func (c *cliEnvOverrides) Set(v string) error {
+ idx := strings.Index(v, "=")
+ if idx <= 0 {
+ return fmt.Errorf("-e expects KEY=VALUE, got %q", v)
+ }
+ c.pairs = append(c.pairs, v)
+ return os.Setenv(v[:idx], v[idx+1:])
+}
diff --git a/core/api/web.go b/core/api/web.go
index 93badfc0a..02b342f72 100755
--- a/core/api/web.go
+++ b/core/api/web.go
@@ -112,6 +112,7 @@ func StartWeb(cfg config.WebAppConfig) {
registerMCPAutoUIHandler(cfg)
+ mountedFuncPatterns := map[string]bool{} // embedding_api 已挂载的 ServeMux 模式
if cfg.EmbeddingAPI {
if registeredAPIMethodHandler != nil {
for k, v := range registeredAPIMethodHandler {
@@ -125,13 +126,19 @@ func StartWeb(cfg config.WebAppConfig) {
for k, v := range registeredAPIFuncHandler {
log.Debug("register http handler: ", k)
uiServeMux.HandleFunc(k, v)
+ mountedFuncPatterns[k] = true
}
}
}
if cfg.WebsocketConfig.Enabled {
websocket.InitWebSocket(cfg.WebsocketConfig)
- uiServeMux.HandleFunc("/ws", websocket.ServeWs)
+ // embedding_api 可能已把 API 域注册的 /ws (StartAPI 的
+ // HandleAPIFunc) 挂到本 mux — 重复 HandleFunc 在 Go 1.22+
+ // 的 ServeMux 语义下 panic, 跳过。
+ if !mountedFuncPatterns["/ws"] && !mountedFuncPatterns[cfg.WebsocketConfig.BasePath] {
+ uiServeMux.HandleFunc("/ws", websocket.ServeWs)
+ }
if registeredWebSocketCommandHandler != nil {
for k, v := range registeredWebSocketCommandHandler {
log.Debug("register websocket handler: ", k, " ", v)
diff --git a/core/api/websocket/conn.go b/core/api/websocket/conn.go
index 56e43c57f..cd33c37c0 100755
--- a/core/api/websocket/conn.go
+++ b/core/api/websocket/conn.go
@@ -30,7 +30,6 @@ package websocket
import (
log "github.com/cihub/seelog"
"github.com/gorilla/websocket"
- "infini.sh/framework/core/global"
"infini.sh/framework/core/util"
"net/http"
"strings"
@@ -48,8 +47,10 @@ const (
// Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10
- // Maximum message size allowed from peer.
- maxMessageSize = 512
+ // Maximum message size allowed from peer. The reverse channel streams
+ // chunked proxied HTTP bodies through this hub (base64 chunks of up to
+ // 32KB), so this must comfortably exceed a chunk frame.
+ maxMessageSize = 8 * 1024 * 1024
)
var upgrader = websocket.Upgrader{
@@ -212,9 +213,7 @@ func ServeWs(w http.ResponseWriter, r *http.Request) {
for _, v := range callbacksOnConnect {
err := v(c.id, w, r)
if err != nil {
- if global.Env().IsDebug {
- log.Error(err)
- }
+ log.Warnf("websocket connection rejected: %v", err)
closeMessage := websocket.FormatCloseMessage(websocket.ClosePolicyViolation, err.Error())
if closeErr := ws.WriteMessage(websocket.CloseMessage, closeMessage); closeErr != nil {
log.Error("Failed to send close message:", closeErr)
diff --git a/core/config/system.go b/core/config/system.go
index 8824e028a..51115a54c 100755
--- a/core/config/system.go
+++ b/core/config/system.go
@@ -302,12 +302,14 @@ type ConfigsConfig struct {
ValidConfigsExtensions []string `config:"valid_config_extensions"`
TLSConfig TLSConfig `config:"tls"` //server or client's certs
ManagerConfig struct {
- LocalConfigsRepoPath string `config:"local_configs_repo_path"`
- BasicAuth BasicAuth `config:"basic_auth"`
+ LocalConfigsRepoPath string `config:"local_configs_repo_path"`
+ BasicAuth BasicAuth `config:"basic_auth"`
+ AccessToken ucfg.SecretString `config:"access_token"`
} `config:"manager"`
- AlwaysRegisterAfterRestart bool `config:"always_register_after_restart"`
- AllowGeneratedMetricsTasks bool `config:"allow_generated_metrics_tasks"`
- IgnoredPath []string `config:"ignored_path"`
+ EnrollmentToken ucfg.SecretString `config:"enrollment_token"` // one-time registration pass (configs.server.enrollment.required)
+ AlwaysRegisterAfterRestart bool `config:"always_register_after_restart"`
+ AllowGeneratedMetricsTasks bool `config:"allow_generated_metrics_tasks"`
+ IgnoredPath []string `config:"ignored_path"`
}
type BasicAuth struct {
diff --git a/core/model/const.go b/core/model/const.go
new file mode 100644
index 000000000..7a9cdca63
--- /dev/null
+++ b/core/model/const.go
@@ -0,0 +1,29 @@
+// Copyright (C) INFINI Labs & INFINI LIMITED.
+//
+// The INFINI Framework is offered under the GNU Affero General Public License v3.0
+// and as commercial software.
+//
+// For commercial licensing, contact us at:
+// - Website: infinilabs.com
+// - Email: hello@infini.ltd
+//
+// Open Source licensed under AGPL V3:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package model
+
+const (
+ CredentialIDSystemKey = "credential_id"
+ API_TOKEN = "X-API-TOKEN"
+)
diff --git a/core/model/instance.go b/core/model/instance.go
index f9c60b44b..59dbcafe3 100644
--- a/core/model/instance.go
+++ b/core/model/instance.go
@@ -59,6 +59,12 @@ type Instance struct {
Labels map[string]string `json:"labels,omitempty" elastic_mapping:"labels:{type:object}"`
Tags []string `json:"tags,omitempty"`
+ // Groups is the SERVER-OWNED grouping of instances (e.g. "es",
+ // "gateway-edge"): managed from the management UI, used to target
+ // config delivery (ManagedConfig.Groups). Instances do not report it —
+ // registration/heartbeat upserts preserve the stored value.
+ Groups []string `json:"groups,omitempty" elastic_mapping:"groups:{type:keyword}}"`
+
//user can pass
Description string `json:"description,omitempty" config:"description" elastic_mapping:"description:{type:keyword}"`
@@ -66,6 +72,11 @@ type Instance struct {
Host *HostInfo `json:"host,omitempty" elastic_mapping:"host: { type: object }"`
+ // AccessToken is the agent's self-generated API token (the console
+ // token-exchange convention): managers store it at registration and
+ // use it for reverse calls (stats, pipeline tasks, proxying).
+ AccessToken *Token `config:"access_token" json:"access_token,omitempty" elastic_mapping:"access_token:{type:object}"`
+
Network NetworkInfo `json:"network,omitempty" elastic_mapping:"network: { type: object }"`
Services []ServiceInfo `json:"services,omitempty" elastic_mapping:"services: { type: object }"`
Status string `json:"status,omitempty" elastic_mapping:"status: { type: keyword, copy_to:search_text }"`
@@ -137,7 +148,20 @@ func GetInstanceInfo() Instance {
_, publicIP, _, _ := util.GetPublishNetworkDeviceInfo(global.Env().SystemConfig.NodeConfig.MajorIpPattern)
- instance.Endpoint = global.Env().SystemConfig.APIConfig.GetEndpoint()
+ // The advertised endpoint must point at a server that actually serves
+ // requests. Deployments commonly disable the dedicated API port
+ // (api.enabled: false) and serve everything on the web port — in that
+ // case advertise the web address, not the (unserving) API default.
+ sysCfg := global.Env().SystemConfig
+ if sysCfg.APIConfig.Enabled {
+ instance.Endpoint = sysCfg.APIConfig.GetEndpoint()
+ } else {
+ schema := "http"
+ if sysCfg.WebAppConfig.TLSConfig.TLSEnabled {
+ schema = "https"
+ }
+ instance.Endpoint = fmt.Sprintf("%s://%s", schema, sysCfg.WebAppConfig.NetworkConfig.GetPublishAddr())
+ }
ips := util.GetLocalIPs()
if len(ips) > 0 {
diff --git a/core/model/token.go b/core/model/token.go
new file mode 100644
index 000000000..b325d90a5
--- /dev/null
+++ b/core/model/token.go
@@ -0,0 +1,9 @@
+/* Copyright © INFINI Ltd. All rights reserved. */
+
+package model
+
+// Token is a bearer credential carried on registered instances (the
+// agent's self-generated API token, stored at registration).
+type Token struct {
+ Value string `json:"value,omitempty" config:"value"`
+}
diff --git a/core/pipeline/record.go b/core/pipeline/record.go
index 78d87721b..b58e35207 100644
--- a/core/pipeline/record.go
+++ b/core/pipeline/record.go
@@ -126,3 +126,33 @@ func AppendFailureTag(ctx *Context, tag string) {
*tags = append(*tags, tag)
}
}
+
+// CloneContextKey is the Context key under which the clone processor
+// deposits cloned records; the batch-splitting host materializes them as
+// additional batch members after the sub-chain finishes the original.
+const CloneContextKey = "record_clones"
+
+// AppendClone registers a clone of the current record for the host to
+// materialize (see clone processor). No-op outside a record scope.
+func AppendClone(ctx *Context, rec *event.Event) {
+ if ctx == nil || rec == nil {
+ return
+ }
+ if list, ok := ctx.Get(CloneContextKey).(*[]*event.Event); ok {
+ *list = append(*list, rec)
+ }
+}
+
+// TakeClones returns and clears the pending clones registered in this
+// context (host-side; call after the per-record sub-chain).
+func TakeClones(ctx *Context) []*event.Event {
+ if ctx == nil {
+ return nil
+ }
+ if list, ok := ctx.Get(CloneContextKey).(*[]*event.Event); ok && len(*list) > 0 {
+ out := *list
+ *list = nil
+ return out
+ }
+ return nil
+}
diff --git a/core/pipeline/register.go b/core/pipeline/register.go
index e7a80ca45..d0c9c99b5 100644
--- a/core/pipeline/register.go
+++ b/core/pipeline/register.go
@@ -251,6 +251,16 @@ type Constructor func(config *config.Config) (ProcessorBase, error)
var registry = NewNamespace()
+// LookupProcessorConstructor resolves a processor constructor by its
+// registered (bare) name — nil when unknown. Hosts use this to build
+// ad-hoc chains (e.g. dry-run replay of a spec's processor list).
+func LookupProcessorConstructor(name string) ProcessorConstructor {
+ if ctor, ok := registry.ProcessorConstructors()[name]; ok {
+ return ctor
+ }
+ return nil
+}
+
func RegisterProcessorPlugin(name string, constructor ProcessorConstructor) {
err := registry.RegisterProcessor(name, constructor)
if err != nil {
diff --git a/modules/configs/client/client.go b/modules/configs/client/client.go
index f863434b2..af662d486 100644
--- a/modules/configs/client/client.go
+++ b/modules/configs/client/client.go
@@ -34,7 +34,9 @@ import (
"net/url"
"os"
"path/filepath"
+ "strings"
"sync"
+ "sync/atomic"
"time"
log "github.com/cihub/seelog"
@@ -46,72 +48,309 @@ import (
"infini.sh/framework/core/model"
"infini.sh/framework/core/task"
"infini.sh/framework/core/util"
+ ucfg "infini.sh/framework/lib/go-ucfg"
"infini.sh/framework/modules/configs/common"
"infini.sh/framework/modules/configs/config"
)
const bucketName = "instance_registered"
const configRegisterEnvKey = "CONFIG_MANAGED_SUCCESS"
+const legacyManagedRegisterCompatMaxVersion = "1.30.4"
+const unauthorizedRegisterRetryInterval = 10 * time.Second
+
+var postRegisterHooks []func(server string, res *util.Result) error
+var unauthorizedRegisterRetryLock sync.Mutex
+var lastUnauthorizedRegisterRetryAt time.Time
+var clearManagedRegistrationStateFunc = clearManagedRegistrationState
+var loadManagedBootstrapAccessTokenFunc = func() (string, error) {
+ return common.LoadTokenFromKeystore(common.ManagerBootstrapTokenKeystoreKey)
+}
+var restoreManagedBootstrapAccessTokenFunc = func() (string, error) {
+ token, err := loadManagedBootstrapAccessTokenFunc()
+ if err != nil {
+ return "", err
+ }
+ token = strings.TrimSpace(token)
+ if token == "" {
+ token = strings.TrimSpace(global.Env().SystemConfig.Configs.ManagerConfig.AccessToken.Get())
+ }
+ if token == "" {
+ token, err = common.LoadTokenFromKeystore(common.ManagerTokenKeystoreKey)
+ if err != nil {
+ return "", err
+ }
+ token = strings.TrimSpace(token)
+ }
+ if token == "" {
+ return "", fmt.Errorf("managed bootstrap access token is missing")
+ }
+ global.Env().SystemConfig.Configs.ManagerConfig.AccessToken = ucfg.SecretString(token)
+ return token, nil
+}
+var reconnectToManagerFunc func() error
+var configSyncInProgress atomic.Bool
-func ConnectToManager() error {
+func init() {
+ reconnectToManagerFunc = ConnectToManager
+}
- if !global.Env().SystemConfig.Configs.Managed {
+// maskURLInError replaces http(s):// URLs in error messages to avoid leaking internal addresses in logs.
+func maskURLInError(err error) string {
+ if err == nil {
+ return ""
+ }
+ msg := err.Error()
+ for _, scheme := range []string{"https://", "http://"} {
+ for {
+ idx := strings.Index(msg, scheme)
+ if idx < 0 {
+ break
+ }
+ end := strings.IndexAny(msg[idx:], " \"'\n\t")
+ if end < 0 {
+ msg = msg[:idx] + "***"
+ break
+ }
+ msg = msg[:idx] + "***" + msg[idx+end:]
+ }
+ }
+ return msg
+}
+
+func truncateManagerResponseBodyForLog(body []byte) string {
+ text := strings.TrimSpace(string(body))
+ if len(text) <= 256 {
+ return text
+ }
+ return text[:256] + "...(truncated)"
+}
+
+func tryStartManagedConfigSync() bool {
+ return configSyncInProgress.CompareAndSwap(false, true)
+}
+
+func finishManagedConfigSync() {
+ configSyncInProgress.Store(false)
+}
+
+func ConnectToManager() error {
+ cfg := global.Env().SystemConfig.Configs
+ if !cfg.Managed {
return nil
}
+ if cfg.Servers == nil || len(cfg.Servers) == 0 {
+ return errors.Errorf("no config manager was found")
+ }
// k8s env setting always_register_after_restart and pod after restart the ip will change so need register again
- if !global.Env().SystemConfig.Configs.AlwaysRegisterAfterRestart {
+ if !cfg.AlwaysRegisterAfterRestart {
if exists, err := kv.ExistsKey(bucketName, []byte(global.Env().SystemConfig.NodeConfig.ID)); exists && err == nil {
//already registered skip further process
- log.Info("already registered to config manager")
+ log.Infof("skip config manager registration for instance %v: local registration marker exists", global.Env().SystemConfig.NodeConfig.ID)
global.Register(configRegisterEnvKey, true)
return nil
}
}
- log.Info("register new instance to config manager")
-
- //register to config manager
- if global.Env().SystemConfig.Configs.Servers == nil || len(global.Env().SystemConfig.Configs.Servers) == 0 {
- return errors.Errorf("no config manager was found")
- }
-
info := model.GetInstanceInfo()
+ log.Infof("start config manager registration for instance %v against %d server(s)", info.ID, len(cfg.Servers))
+ registerReq := common.InstanceRegisterRequest{
+ Client: info,
+ }
+ registerAccessToken, err := buildManagedRegisterAccessToken(info)
+ if err != nil {
+ return err
+ }
+ if registerAccessToken != nil {
+ registerReq.AccessToken = registerAccessToken
+ }
req := util.Request{Method: util.Verb_POST}
req.ContentType = "application/json"
req.Path = common.REGISTER_API
- req.Body = util.MustToJSONBytes(info)
+ req.Body = util.MustToJSONBytes(registerReq)
+ // Enrollment ticket (one-time registration pass; required when the
+ // server sets configs.server.enrollment.required: true).
+ if et := global.Env().SystemConfig.Configs.EnrollmentToken.Get(); et != "" {
+ req.AddHeader("X-Enrollment-Token", et)
+ }
server, res, err := submitRequestToManager(&req)
if err == nil && server != "" {
if res.StatusCode == 200 || util.ContainStr(string(res.Body), "exists") {
- log.Infof("success register to config manager: %v", string(server))
+ // Admission flow: the manager may hold the instance as PENDING
+ // (visible, no credentials). Capture the manager token when
+ // present; otherwise stay unregistered so we keep re-registering
+ // until an admin approves us.
+ var regResp struct {
+ Approved bool `json:"approved"`
+ ManagerToken string `json:"manager_token"`
+ }
+ if util.FromJSONBytes(res.Body, ®Resp) == nil && regResp.ManagerToken != "" {
+ _ = keystore.SetValue(common.ManagerTokenKeystoreKey, []byte(regResp.ManagerToken))
+ global.Env().SystemConfig.Configs.ManagerConfig.AccessToken = ucfg.SecretString(regResp.ManagerToken)
+ log.Info("received manager token from config manager")
+ }
+ if util.FromJSONBytes(res.Body, ®Resp) == nil && !regResp.Approved {
+ log.Warnf("instance %v registered as PENDING on %v - waiting for admin approval", info.ID, server)
+ return nil // no local marker: re-register next cycle until approved
+ }
+ if err := execPostRegisterHooks(server, res); err != nil {
+ return err
+ }
+ log.Infof("config manager registration succeeded for instance %v via %v: status=%d", info.ID, server, res.StatusCode)
err := kv.AddValue(bucketName, []byte(global.Env().SystemConfig.NodeConfig.ID), []byte(util.GetLowPrecisionCurrentTime().String()))
if err != nil {
panic(err)
}
global.Register(configRegisterEnvKey, true)
+ } else {
+ if res.StatusCode == http.StatusUnauthorized {
+ if !claimUnauthorizedRegisterRetrySlot() {
+ return fmt.Errorf("unauthorized config manager registration")
+ }
+ return recoverManagedRegistrationWithBootstrap()
+ }
+ log.Warnf("config manager registration failed for instance %v via %v: status=%d, body=%s", info.ID, server, res.StatusCode, truncateManagerResponseBodyForLog(res.Body))
+ return fmt.Errorf("failed to register to config manager: status=%d, body=%s", res.StatusCode, strings.TrimSpace(string(res.Body)))
}
} else {
- log.Error("failed to register to config manager,", err, ",", server)
+ log.Errorf("config manager registration request failed for instance %v via %v: %v", info.ID, server, err)
}
return err
}
+func buildManagedRegisterAccessToken(info model.Instance) (*common.RegisterToken, error) {
+ if !common.SupportsManagedAccessToken(info.Application.Name) {
+ return nil, nil
+ }
+ if shouldSkipManagedRegisterAccessToken(info.Application.Version.VersionNumber) {
+ return nil, nil
+ }
+ accessToken, err := common.EnsureTokenInKeystore(common.InstanceAccessTokenKeystoreKey)
+ if err != nil {
+ return nil, err
+ }
+ productName := strings.TrimSpace(info.Application.Name)
+ if productName == "" {
+ productName = "instance"
+ }
+ return &common.RegisterToken{
+ Name: fmt.Sprintf("%s access token", info.ID),
+ Description: fmt.Sprintf("Console to %s access token for instance %s", productName, info.ID),
+ Value: accessToken,
+ }, nil
+}
+
+func shouldSkipManagedRegisterAccessToken(version string) bool {
+ version = strings.TrimSpace(version)
+ if version == "" {
+ return false
+ }
+ // Snapshot/dev builds (e.g. 0.0.1-SNAPSHOT) carry the NEWEST code —
+ // they must not be classified as legacy by their low version number.
+ for _, marker := range []string{"SNAPSHOT", "snapshot", "dev", "DEV"} {
+ if strings.Contains(version, marker) {
+ return false
+ }
+ }
+ parsed, err := util.ParseSemantic(version)
+ if err != nil {
+ parsed, err = util.ParseGeneric(version)
+ if err != nil {
+ return false
+ }
+ }
+ cmp, err := parsed.Compare(legacyManagedRegisterCompatMaxVersion)
+ if err != nil {
+ return false
+ }
+ return cmp <= 0
+}
+
+func AddPostRegisterHook(hook func(server string, res *util.Result) error) {
+ if hook != nil {
+ postRegisterHooks = append(postRegisterHooks, hook)
+ }
+}
+
+func clearManagedRegistrationState() error {
+ global.Register(configRegisterEnvKey, false)
+ instanceID := strings.TrimSpace(global.Env().SystemConfig.NodeConfig.ID)
+ if instanceID == "" {
+ return nil
+ }
+ return kv.DeleteKey(bucketName, []byte(instanceID))
+}
+
+func handleUnauthorizedConfigSyncResponse(res *util.Result) bool {
+ if res == nil || res.StatusCode != http.StatusUnauthorized {
+ return false
+ }
+
+ if !claimUnauthorizedRegisterRetrySlot() {
+ return true
+ }
+
+ log.Warn("config sync unauthorized, clearing local registration state and retrying registration")
+ if err := recoverManagedRegistrationWithBootstrap(); err != nil {
+ log.Warnf("failed to re-register to config manager after unauthorized config sync: %v", err)
+ return true
+ }
+ log.Info("re-registered to config manager after unauthorized config sync")
+ return true
+}
+
+func claimUnauthorizedRegisterRetrySlot() bool {
+ unauthorizedRegisterRetryLock.Lock()
+ defer unauthorizedRegisterRetryLock.Unlock()
+ if !lastUnauthorizedRegisterRetryAt.IsZero() && time.Since(lastUnauthorizedRegisterRetryAt) < unauthorizedRegisterRetryInterval {
+ return false
+ }
+ lastUnauthorizedRegisterRetryAt = time.Now()
+ return true
+}
+
+func recoverManagedRegistrationWithBootstrap() error {
+ if err := clearManagedRegistrationStateFunc(); err != nil {
+ return err
+ }
+ // Bootstrap token is OPTIONAL: admission-mode servers approve manually
+ // and issue credentials on approve; retrying registration plainly is
+ // the correct recovery there. Only static-token servers need the
+ // bootstrap restore, and its absence is not an error.
+ if _, err := restoreManagedBootstrapAccessTokenFunc(); err != nil {
+ log.Debugf("no bootstrap token configured (admission-mode server?): %v", err)
+ }
+ return reconnectToManagerFunc()
+}
+
+func execPostRegisterHooks(server string, res *util.Result) error {
+ for _, hook := range postRegisterHooks {
+ if err := hook(server, res); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
func submitRequestToManager(req *util.Request) (string, *util.Result, error) {
+ return DoManagerRequest(req)
+}
+
+func DoManagerRequest(req *util.Request) (string, *util.Result, error) {
var err error
var res *util.Result
cfg := global.Env().SystemConfig.Configs
- if cfg.ManagerConfig.BasicAuth.Username != "" {
- req.SetBasicAuth(cfg.ManagerConfig.BasicAuth.Username, cfg.ManagerConfig.BasicAuth.Password.Get())
+ if err = applyManagerRequestAuth(req); err != nil {
+ return "", nil, err
}
for _, server := range cfg.Servers {
req.Url, err = url.JoinPath(server, req.Path)
if err != nil {
continue
}
- res, err = util.ExecuteRequestWithCatchFlag(mTLSClient, req, true)
+ res, err = util.ExecuteRequestWithCatchFlag(getManagerHTTPClient(), req, true)
if err != nil {
continue
}
@@ -120,32 +359,72 @@ func submitRequestToManager(req *util.Request) (string, *util.Result, error) {
return "", nil, err
}
-var clientInitLock = sync.Once{}
+func applyManagerRequestAuth(req *util.Request) error {
+ cfg := global.Env().SystemConfig.Configs
+ if token := cfg.ManagerConfig.AccessToken.Get(); token != "" {
+ req.AddHeader(model.API_TOKEN, token)
+ return nil
+ }
+ token, err := common.LoadTokenFromKeystore(common.ManagerTokenKeystoreKey)
+ if err != nil {
+ return err
+ }
+ if token != "" {
+ req.AddHeader("Authorization", "Bearer "+token)
+ return nil
+ }
+ if cfg.ManagerConfig.BasicAuth.Username != "" {
+ req.SetBasicAuth(cfg.ManagerConfig.BasicAuth.Username, cfg.ManagerConfig.BasicAuth.Password.Get())
+ }
+ return nil
+}
+
+var managerHTTPClientInitLock = sync.Once{}
+var configSyncInitLock = sync.Once{}
var mTLSClient *http.Client
-func ListenConfigChanges() error {
+func initManagerHTTPClient() {
+ managerHTTPClientInitLock.Do(func() {
+ if !global.Env().SystemConfig.Configs.Managed {
+ return
+ }
+ cfg := global.Env().GetHTTPClientConfig("configs", "")
+ if cfg != nil {
+ hClient, err := api.NewHTTPClient(cfg)
+ if err != nil {
+ panic(err)
+ }
+ mTLSClient = hClient
+ }
+ })
+}
- clientInitLock.Do(func() {
+func getManagerHTTPClient() *http.Client {
+ initManagerHTTPClient()
+ return mTLSClient
+}
- if global.Env().SystemConfig.Configs.Managed {
- cfg := global.Env().GetHTTPClientConfig("configs", "")
- if cfg != nil {
- hClient, err := api.NewHTTPClient(cfg)
- if err != nil {
- panic(err)
- }
- mTLSClient = hClient
- }
+func ListenConfigChanges() error {
+ configSyncInitLock.Do(func() {
- //init config sync listening
- req := common.ConfigSyncRequest{}
- req.Client = model.GetInstanceInfo()
+ if global.Env().SystemConfig.Configs.Managed {
+ initManagerHTTPClient()
var syncFunc = func() {
+ if !tryStartManagedConfigSync() {
+ if global.Env().IsDebug {
+ log.Trace("skip overlapping config sync")
+ }
+ return
+ }
+ defer finishManagedConfigSync()
+
if global.Env().IsDebug {
log.Trace("fetch configs from manger")
}
+ req := common.ConfigSyncRequest{}
+ req.Client = model.GetInstanceInfo()
cfgs := config.GetConfigs(false, false)
req.Configs = cfgs
req.Hash = util.MD5digestString(util.MustToJSONBytes(cfgs))
@@ -154,19 +433,24 @@ func ListenConfigChanges() error {
request := util.Request{Method: util.Verb_POST}
request.ContentType = "application/json"
request.Path = common.SYNC_API
- request.Body = util.MustToJSONBytes(req)
+ requestBody := util.MustToJSONBytes(req)
+ request.Body = requestBody
if global.Env().IsDebug {
- log.Debug("config sync request: ", string(util.MustToJSONBytes(req)))
+ log.Debug("config sync request: ", string(requestBody))
}
- _, res, err := submitRequestToManager(&request)
+ _, res, err := DoManagerRequest(&request)
if err != nil {
- log.Error("failed to submit request to config manager,", err)
+ log.Error("failed to submit request to config manager,", maskURLInError(err))
return
}
if res != nil {
+ if handleUnauthorizedConfigSyncResponse(res) {
+ return
+ }
+
obj := common.ConfigSyncResponse{}
err := util.FromJSONBytes(res.Body, &obj)
if err != nil {
diff --git a/modules/configs/common/config.go b/modules/configs/common/config.go
index 03ce8dbe4..a13c8675e 100644
--- a/modules/configs/common/config.go
+++ b/modules/configs/common/config.go
@@ -38,11 +38,12 @@ type AgentConfig struct {
}
type SetupConfig struct {
- DownloadURL string `config:"download_url"`
- CACertFile string `config:"ca_cert"`
- CAKeyFile string `config:"ca_key"`
- ConsoleEndpoint string `config:"console_endpoint"`
- Port string `config:"port"`
+ DownloadURL string `config:"download_url"`
+ CACertFile string `config:"ca_cert"`
+ CAKeyFile string `config:"ca_key"`
+ ConsoleEndpoint string `config:"console_endpoint"`
+ ReverseChannelEndpoints []string `config:"reverse_channel_endpoints"`
+ Port string `config:"port"`
}
func GetAgentConfig() *AgentConfig {
diff --git a/modules/configs/common/domain.go b/modules/configs/common/domain.go
index 232c1fc52..46ecbbcaf 100644
--- a/modules/configs/common/domain.go
+++ b/modules/configs/common/domain.go
@@ -27,11 +27,32 @@
package common
-import "infini.sh/framework/core/model"
+import (
+ "strings"
+
+ "infini.sh/framework/core/model"
+)
const REGISTER_API = "/instance/_register"
const SYNC_API = "/configs/_sync"
+const (
+ ManagerTokenKeystoreKey = "configs_manager_token"
+ ManagerBootstrapTokenKeystoreKey = "configs_manager_bootstrap_token"
+ InstanceAccessTokenKeystoreKey = "instance_access_token"
+)
+
+type RegisterToken struct {
+ Name string `json:"name,omitempty"`
+ Description string `json:"description,omitempty"`
+ Value string `json:"value,omitempty"`
+}
+
+type InstanceRegisterRequest struct {
+ Client model.Instance `json:"client"`
+ AccessToken *RegisterToken `json:"access_token,omitempty"`
+}
+
type ConfigFile struct {
Name string `json:"name,omitempty"`
Location string `json:"location,omitempty"`
@@ -109,3 +130,12 @@ type InstanceSettings struct {
ConfigFiles []string `config:"configs"`
Secrets []string `config:"secrets"`
}
+
+// SupportsManagedAccessToken reports whether an application joining the
+// managed-config flow should hand a keystore-backed access token to the
+// console at registration time. Any Framework-based instance (Agent,
+// Gateway, or third-party apps) qualifies; empty names are skipped since
+// there is nothing meaningful to mint a token for.
+func SupportsManagedAccessToken(applicationName string) bool {
+ return strings.TrimSpace(applicationName) != ""
+}
diff --git a/modules/configs/common/token.go b/modules/configs/common/token.go
new file mode 100644
index 000000000..7f7e634cc
--- /dev/null
+++ b/modules/configs/common/token.go
@@ -0,0 +1,66 @@
+// Copyright (C) INFINI Labs & INFINI LIMITED.
+//
+// The INFINI Framework is offered under the GNU Affero General Public License v3.0
+// and as commercial software.
+//
+// For commercial licensing, contact us at:
+// - Website: infinilabs.com
+// - Email: hello@infini.ltd
+//
+// Open Source licensed under AGPL V3:
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+/* Copyright © INFINI LTD. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package common
+
+import (
+ "strings"
+
+ "infini.sh/framework/core/keystore"
+ "infini.sh/framework/core/util"
+ keystore2 "infini.sh/framework/lib/keystore"
+)
+
+func LoadTokenFromKeystore(key string) (string, error) {
+ value, err := keystore.GetValue(key)
+ if err == keystore2.ErrKeyDoesntExists {
+ return "", nil
+ }
+ if err != nil {
+ return "", err
+ }
+ return strings.TrimSpace(string(value)), nil
+}
+
+func SaveTokenToKeystore(key, value string) error {
+ return keystore.SetValue(key, util.UnsafeStringToBytes(strings.TrimSpace(value)))
+}
+
+func EnsureTokenInKeystore(key string) (string, error) {
+ value, err := LoadTokenFromKeystore(key)
+ if err != nil {
+ return "", err
+ }
+ if value != "" {
+ return value, nil
+ }
+ value = util.GenerateRandomString(48)
+ if err := SaveTokenToKeystore(key, value); err != nil {
+ return "", err
+ }
+ return value, nil
+}
diff --git a/modules/configs/reverseclient/dispatch_test.go b/modules/configs/reverseclient/dispatch_test.go
new file mode 100644
index 000000000..7a76a79dc
--- /dev/null
+++ b/modules/configs/reverseclient/dispatch_test.go
@@ -0,0 +1,47 @@
+/* Copyright © INFINI Ltd. All rights reserved. */
+
+package reverse
+
+import (
+ "strings"
+ "testing"
+)
+
+// dispatch mimics serve()'s frame dispatch; kept in sync manually.
+func dispatch(frame string) (command, payload string) {
+ parts := strings.SplitN(frame, " ", 2)
+ if len(parts) != 2 {
+ return "", ""
+ }
+ payload = parts[1]
+ command = parts[0]
+ if command == "PRIVATE" || command == "CONFIG" {
+ if sub := strings.SplitN(payload, " ", 2); len(sub) == 2 && strings.HasPrefix(sub[0], "reverse_") {
+ command, payload = sub[0], sub[1]
+ }
+ }
+ return command, payload
+}
+
+func TestFrameDispatch(t *testing.T) {
+ cases := []struct {
+ frame string
+ command string
+ payload string
+ }{
+ {"CONFIG websocket-session-id: abc", "CONFIG", "websocket-session-id: abc"},
+ {`PRIVATE reverse_request {"id":"1"}`, "reverse_request", `{"id":"1"}`},
+ {"reverse_request {}", "reverse_request", "{}"},
+ {"PRIVATE something_else x", "PRIVATE", "something_else x"},
+ {"", "", ""},
+ }
+ for _, c := range cases {
+ command, payload := dispatch(c.frame)
+ if command != c.command {
+ t.Errorf("frame %q: command = %q, want %q", c.frame, command, c.command)
+ }
+ if payload != c.payload {
+ t.Errorf("frame %q: payload = %q, want %q", c.frame, payload, c.payload)
+ }
+ }
+}
diff --git a/modules/configs/reverseclient/module.go b/modules/configs/reverseclient/module.go
new file mode 100644
index 000000000..7525fccdc
--- /dev/null
+++ b/modules/configs/reverseclient/module.go
@@ -0,0 +1,20 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package reverse
+
+import "infini.sh/framework/core/module"
+
+type Module struct{}
+
+func (m *Module) Name() string { return "reverse_channel" }
+func (m *Module) Setup() {}
+func (m *Module) Start() error { Setup(); return nil }
+func (m *Module) Stop() error { return nil }
+
+func init() {
+ // After the managed bootstrap (100) so the manager token from
+ // registration/exchange is likely already in the keystore.
+ module.RegisterModuleWithPriority(&Module{}, 101)
+}
diff --git a/modules/configs/reverseclient/reverse.go b/modules/configs/reverseclient/reverse.go
new file mode 100644
index 000000000..e31fc23ec
--- /dev/null
+++ b/modules/configs/reverseclient/reverse.go
@@ -0,0 +1,277 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package reverse implements the agent side of the reverse channel: the
+// agent — typically behind NAT/firewall and NOT directly reachable —
+// dials OUT to the config manager's websocket endpoint and then serves
+// the manager's HTTP requests THROUGH that connection (executed as
+// loopback calls against the agent's own web port).
+package reverse
+
+import (
+ "encoding/base64"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+ "sync"
+ "time"
+
+ log "github.com/cihub/seelog"
+ "github.com/gorilla/websocket"
+
+ "infini.sh/framework/core/api/websocket/reverse"
+ "infini.sh/framework/core/global"
+ "infini.sh/framework/core/util"
+ common "infini.sh/framework/modules/configs/common"
+)
+
+const (
+ reconnectDelay = 5 * time.Second
+ maxMessageBytes = 8 * 1024 * 1024
+ requestDeadline = 30 * time.Second
+)
+
+var (
+ writeMu sync.Mutex
+ started sync.Once
+)
+
+// Setup launches the reverse-channel loop (idempotent). The agent main
+// calls this; it only dials out when configs.managed is on.
+func Setup() {
+ started.Do(func() {
+ if !global.Env().SystemConfig.Configs.Managed {
+ return
+ }
+ go run()
+ })
+}
+
+func run() {
+ for !global.ShuttingDown() {
+ if err := connectAndServe(); err != nil && !global.ShuttingDown() {
+ log.Debugf("agent reverse channel: %v (retrying in %v)", err, reconnectDelay)
+ }
+ if global.ShuttingDown() {
+ return
+ }
+ time.Sleep(reconnectDelay)
+ }
+}
+
+func connectAndServe() error {
+ servers := global.Env().SystemConfig.Configs.Servers
+ var lastErr error
+ for _, server := range servers {
+ conn, err := dial(server)
+ if err != nil {
+ lastErr = err
+ continue
+ }
+ log.Infof("agent reverse channel connected to [%s]", server)
+ err = serve(conn)
+ _ = conn.Close()
+ log.Warnf("agent reverse channel disconnected from [%s]: %v", server, err)
+ return err
+ }
+ if lastErr != nil {
+ return lastErr
+ }
+ return nil
+}
+
+// dial opens the websocket to the manager's /ws endpoint, carrying the
+// instance ID and the manager credential (same auth as sync).
+func dial(server string) (*websocket.Conn, error) {
+ wsURL, err := reverseURL(server)
+ if err != nil {
+ return nil, err
+ }
+ headers := http.Header{}
+ headers.Set(reverse.HeaderPeerID, global.Env().SystemConfig.NodeConfig.ID)
+ if tok := global.Env().SystemConfig.Configs.ManagerConfig.AccessToken.Get(); tok != "" {
+ headers.Set("Authorization", "Bearer "+tok)
+ } else if tok, _ := common.LoadTokenFromKeystore(common.ManagerTokenKeystoreKey); tok != "" {
+ headers.Set("Authorization", "Bearer "+strings.TrimSpace(tok))
+ }
+ dialer := &websocket.Dialer{HandshakeTimeout: 10 * time.Second}
+ conn, _, err := dialer.Dial(wsURL, headers)
+ return conn, err
+}
+
+func reverseURL(server string) (string, error) {
+ u, err := url.Parse(server)
+ if err != nil {
+ return "", err
+ }
+ switch u.Scheme {
+ case "https":
+ u.Scheme = "wss"
+ default:
+ u.Scheme = "ws"
+ }
+ if !strings.HasSuffix(u.Path, "/ws") {
+ u.Path = strings.TrimSuffix(u.Path, "/") + "/ws"
+ }
+ return u.String(), nil
+}
+
+// serve reads frames: session assignment → hello → request loop.
+func serve(conn *websocket.Conn) error {
+ conn.SetReadLimit(maxMessageBytes)
+ for {
+ _, payload, err := conn.ReadMessage()
+ if err != nil {
+ return err
+ }
+ text := string(payload)
+ // Hub wire format is " "; the proxied requests
+ // arrive as "PRIVATE reverse_request {json}" — strip the type
+ // prefix first, then dispatch on the command.
+ parts := strings.SplitN(text, " ", 2)
+ if len(parts) != 2 {
+ continue
+ }
+ payload1 := parts[1]
+ command := parts[0]
+ if command == "PRIVATE" || command == "CONFIG" {
+ // "PRIVATE reverse_request {json}" → command=reverse_request
+ if sub := strings.SplitN(payload1, " ", 2); len(sub) == 2 && strings.HasPrefix(sub[0], "reverse_") {
+ command, payload1 = sub[0], sub[1]
+ }
+ }
+ switch command {
+ case "CONFIG":
+ if sid, ok := stripPrefix(payload1, "websocket-session-id:"); ok && sid != "" {
+ hello := reverse.HelloMessage{
+ SessionID: sid,
+ PeerID: global.Env().SystemConfig.NodeConfig.ID,
+ }
+ if err := send(conn, reverse.FormatHelloCommand(hello)); err != nil {
+ return err
+ }
+ log.Debugf("agent reverse channel hello sent for session [%s]", sid)
+ }
+ case reverse.RequestCommand:
+ go handleRequest(conn, payload1)
+ }
+ }
+}
+
+func stripPrefix(s, prefix string) (string, bool) {
+ if strings.HasPrefix(s, prefix) {
+ return strings.TrimSpace(s[len(prefix):]), true
+ }
+ return "", false
+}
+
+func send(conn *websocket.Conn, payload string) error {
+ writeMu.Lock()
+ defer writeMu.Unlock()
+ return conn.WriteMessage(websocket.TextMessage, []byte(payload))
+}
+
+// handleRequest executes one proxied HTTP request against the agent's
+// own web port and streams the response back in chunks.
+func handleRequest(conn *websocket.Conn, payload string) {
+ reqMsg, err := reverse.ParseRequestPayload(payload)
+ if err != nil {
+ log.Debugf("agent reverse channel: bad request payload: %v", err)
+ return
+ }
+
+ status, body := execute(reqMsg)
+
+ resp := reverse.ResponseMessage{
+ RequestID: reqMsg.RequestID,
+ PeerID: reqMsg.PeerID,
+ }
+ // chunk the body (base64) then a Done frame with the status
+ for offset := 0; offset < len(body); offset += 64 * 1024 {
+ end := offset + 64*1024
+ if end > len(body) {
+ end = len(body)
+ }
+ resp.Chunk = base64.StdEncoding.EncodeToString(body[offset:end])
+ if err := send(conn, reverse.FormatResponseCommand(resp)); err != nil {
+ log.Debugf("agent reverse channel: response chunk failed: %v", err)
+ return
+ }
+ }
+ resp.Chunk, resp.Done, resp.Status = "", true, status
+ if len(body) == 0 {
+ // still send one empty chunk so the manager assembles something
+ resp.Chunk = ""
+ }
+ if err := send(conn, reverse.FormatResponseCommand(resp)); err != nil {
+ log.Debugf("agent reverse channel: response done failed: %v", err)
+ }
+}
+
+// execute performs the loopback HTTP call against the agent's web port.
+func execute(reqMsg reverse.RequestMessage) (int, []byte) {
+ target := localBaseURL() + reqMsg.Path
+ var body io.Reader
+ if b, err := reqMsg.BodyBytes(); err == nil && len(b) > 0 {
+ body = strings.NewReader(string(b))
+ }
+ req, err := http.NewRequest(reqMsg.Method, target, body)
+ if err != nil {
+ return http.StatusBadRequest, []byte(err.Error())
+ }
+ reqMsg.ApplyHeaders(req)
+ if req.Header.Get("Content-Type") == "" {
+ req.Header.Set("Content-Type", util.ContentTypeJson)
+ }
+ // Authenticate the loopback call with the agent's own API token so
+ // it passes the access_token realm.
+ if tok, _ := common.LoadTokenFromKeystore("AGENT_API_ACCESS_TOKEN"); tok != "" {
+ req.Header.Set("X-API-Token", strings.TrimSpace(tok))
+ }
+ client := &http.Client{Timeout: requestDeadline}
+ resp, err := client.Do(req)
+ if err != nil {
+ return http.StatusBadGateway, []byte(err.Error())
+ }
+ defer func() { _ = resp.Body.Close() }()
+ data, _ := io.ReadAll(resp.Body)
+ return resp.StatusCode, data
+}
+
+// localBaseURL derives the agent's own serving address.
+func localBaseURL() string {
+ // Prefer the web app port; some binaries (gateway) serve their API
+ // under the `api:` section instead with an empty WebAppConfig network
+ // — GetBindingAddr panics on that, so fall through to APIConfig.
+ var schema, addr string
+ web := global.Env().SystemConfig.WebAppConfig
+ if web.NetworkConfig.Port > 0 || web.NetworkConfig.Binding != "" || web.NetworkConfig.Host != "" {
+ schema = "http"
+ if web.TLSConfig.TLSEnabled {
+ schema = "https"
+ }
+ addr = web.NetworkConfig.GetBindingAddr()
+ } else if api := global.Env().SystemConfig.APIConfig; api.Enabled && (api.NetworkConfig.Port > 0 || api.NetworkConfig.Binding != "" || api.NetworkConfig.Host != "") {
+ schema = "http"
+ if api.TLSConfig.TLSEnabled {
+ schema = "https"
+ }
+ addr = api.NetworkConfig.GetBindingAddr()
+ } else {
+ return "http://127.0.0.1"
+ }
+ // Loopback preference: wildcard bindings (0.0.0.0/::) are reached via
+ // 127.0.0.1, but a binding pinned to a specific interface IP (e.g. a
+ // second instance on a LAN address) must keep that host — nothing
+ // listens on the loopback rewrite for such bindings.
+ if i := strings.LastIndex(addr, ":"); i > 0 {
+ host := addr[:i]
+ if host == "" || host == "0.0.0.0" || host == "::" {
+ host = "127.0.0.1"
+ }
+ return schema + "://" + host + addr[i:]
+ }
+ return schema + "://127.0.0.1"
+}
diff --git a/modules/configs/server/enrollment_token.go b/modules/configs/server/enrollment_token.go
new file mode 100644
index 000000000..3ef9374c9
--- /dev/null
+++ b/modules/configs/server/enrollment_token.go
@@ -0,0 +1,301 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "crypto/rand"
+ "encoding/hex"
+ "net/http"
+ "strings"
+ "sync"
+ "time"
+
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
+ log "infini.sh/framework/core/log"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Enrollment tokens — one-time (or limited-use) admission tickets.
+//
+// The register endpoint is reachable by design (new agents have no
+// credentials yet), which makes it a spam/forgery surface. An enrollment
+// token is the out-of-band distributed ticket that redeems registration:
+//
+// admin generates (UI/API, TTL + max uses) → embeds in the agent's
+// config (configs.enrollment_token) → agent registers presenting it →
+// server validates (hash, expiry, remaining uses) and consumes it.
+//
+// Forged/flooded registrations without a valid ticket are rejected with
+// 403 before any record is written. Combined with the admission flow
+// (pending → approve), this closes the register surface completely:
+// no ticket → no pending record; no approval → no credentials/configs.
+//
+// Storage: sha256-at-rest like instance tokens; the plaintext is shown
+// exactly once, at generation time.
+// ──────────────────────────────────────────────────────────────────────────
+
+// EnrollmentTokenKeystorePrefix distinguishes enrollment tokens in storage.
+const enrollmentTokenModel = "enrollment-tokens"
+
+// EnrollmentToken is a limited-use registration ticket.
+type EnrollmentToken struct {
+ orm.ORMObjectBase
+
+ Name string `json:"name,omitempty"`
+
+ // TokenHash is sha256(plaintext); the plaintext exists only in the
+ // generation response.
+ TokenHash string `json:"token_hash" elastic_mapping:"token_hash:{type:keyword}"`
+
+ // MaxUses caps redemptions (1 = one-time). 0 treated as 1.
+ MaxUses int `json:"max_uses" elastic_mapping:"max_uses:{type:integer}"`
+
+ // UsedCounts is the redemption counter.
+ UsedCount int `json:"used_count" elastic_mapping:"used_count:{type:integer}"`
+
+ // ExpiresAt: zero value = no expiry (still bounded by MaxUses).
+ ExpiresAt time.Time `json:"expires_at,omitempty" elastic_mapping:"expires_at:{type:date}"`
+
+ // Revoked soft-deletes the token.
+ Revoked bool `json:"revoked,omitempty" elastic_mapping:"revoked:{type:boolean}"`
+
+ CreatedBy string `json:"created_by,omitempty"`
+ CreatedAt time.Time `json:"created_at"`
+}
+
+const enrollmentTokenPrefix = "et-"
+
+// mintEnrollmentToken creates a token with the given policy and returns
+// (record, plaintext). The plaintext is shown to the operator exactly once.
+func mintEnrollmentToken(name string, maxUses int, ttl time.Duration, createdBy string) (*EnrollmentToken, string, error) {
+ raw := make([]byte, 24)
+ if _, err := rand.Read(raw); err != nil {
+ return nil, "", err
+ }
+ plaintext := enrollmentTokenPrefix + hex.EncodeToString(raw)
+
+ if maxUses < 1 {
+ maxUses = 1
+ }
+ rec := &EnrollmentToken{
+ Name: name,
+ TokenHash: hashToken(plaintext),
+ MaxUses: maxUses,
+ ExpiresAt: time.Now().UTC().Add(ttl),
+ CreatedBy: createdBy,
+ CreatedAt: time.Now().UTC(),
+ }
+ rec.ID = util.GetUUID()
+ return rec, plaintext, nil
+}
+
+// validateEnrollmentToken checks a presented enrollment ticket WITHOUT
+// consuming it. Returns the record when valid (known, unrevoked,
+// unexpired, uses remaining), nil otherwise.
+func validateEnrollmentToken(ctx *orm.Context, plaintext string) *EnrollmentToken {
+ if plaintext == "" || !strings.HasPrefix(plaintext, enrollmentTokenPrefix) {
+ return nil
+ }
+ orm.WithModel(ctx, &EnrollmentToken{})
+ qb := orm.NewQuery().
+ Filter(orm.TermQuery("token_hash", hashToken(plaintext))).
+ Size(1)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil || res == nil {
+ return nil
+ }
+ tokens, _, _ := decodeEnrollmentHits(res)
+ if len(tokens) == 0 {
+ return nil
+ }
+ t := &tokens[0]
+ if t.Revoked || t.UsedCount >= t.MaxUses {
+ return nil
+ }
+ if !t.ExpiresAt.IsZero() && time.Now().UTC().After(t.ExpiresAt) {
+ return nil
+ }
+ return t
+}
+
+// consumeEnrollmentToken increments the ticket's use counter. Called only
+// when a registration actually CREATES a new instance — re-registration
+// by a known instance must not burn uses (pending agents re-register
+// every sync cycle while awaiting approval).
+func consumeEnrollmentToken(ctx *orm.Context, t *EnrollmentToken) {
+ if t == nil {
+ return
+ }
+ t.UsedCount++
+ if err := orm.Save(ctx, t); err != nil {
+ log.Warnf("configs server: failed to consume enrollment token %s: %v", t.ID, err)
+ }
+}
+
+func decodeEnrollmentHits(res *orm.SearchResult) ([]EnrollmentToken, int64, error) {
+ return elastic.DecodeHits[EnrollmentToken](res)
+}
+
+// rateLimiter is a small fixed-window per-key limiter for the register
+// endpoint (flood control on the publicly reachable surface).
+type rateLimiter struct {
+ mu sync.Mutex
+ window time.Duration
+ maxHits int
+ hits map[string]*hitBucket
+}
+
+type hitBucket struct {
+ count int
+ since time.Time
+}
+
+func newRateLimiter(window time.Duration, maxHits int) *rateLimiter {
+ return &rateLimiter{window: window, maxHits: maxHits, hits: map[string]*hitBucket{}}
+}
+
+func (r *rateLimiter) allow(key string) bool {
+ if r == nil || r.maxHits <= 0 {
+ return true
+ }
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ now := time.Now()
+ b, ok := r.hits[key]
+ if !ok || now.Sub(b.since) >= r.window {
+ // lazy reset; also prune stale entries opportunistically
+ if len(r.hits) > 4096 {
+ r.hits = map[string]*hitBucket{}
+ }
+ r.hits[key] = &hitBucket{count: 1, since: now}
+ return true
+ }
+ b.count++
+ return b.count <= r.maxHits
+}
+
+func clientIP(req *http.Request) string {
+ // best-effort: proxy chains first, then remote addr
+ if fwd := req.Header.Get("X-Forwarded-For"); fwd != "" {
+ return strings.TrimSpace(strings.Split(fwd, ",")[0])
+ }
+ if real := req.Header.Get("X-Real-IP"); real != "" {
+ return real
+ }
+ addr := req.RemoteAddr
+ if i := strings.LastIndex(addr, ":"); i > 0 {
+ addr = addr[:i]
+ }
+ return addr
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// HTTP: management endpoints for enrollment tokens (admin-gated by the
+// static token gate, same as _approve).
+// ──────────────────────────────────────────────────────────────────────────
+
+// enrollmentTokensHandler — GET /instance/_enrollment_tokens (list, masked)
+func (h *APIHandler) enrollmentTokensHandler(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ orm.WithModel(ctx, &EnrollmentToken{})
+ qb := orm.NewQuery().
+ SortBy(orm.Sort{Field: "created_at", SortType: orm.DESC}).
+ Size(100)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil || res == nil {
+ h.WriteJSON(w, util.MapStr{"tokens": []interface{}{}}, http.StatusOK)
+ return
+ }
+ tokens, _, _ := decodeEnrollmentHits(res)
+ out := make([]util.MapStr, 0, len(tokens))
+ for _, t := range tokens {
+ status := "valid"
+ if t.Revoked {
+ status = "revoked"
+ } else if t.UsedCount >= t.MaxUses {
+ status = "exhausted"
+ } else if !t.ExpiresAt.IsZero() && time.Now().UTC().After(t.ExpiresAt) {
+ status = "expired"
+ }
+ out = append(out, util.MapStr{
+ "id": t.ID,
+ "name": t.Name,
+ "max_uses": t.MaxUses,
+ "used_count": t.UsedCount,
+ "expires_at": t.ExpiresAt,
+ "status": status,
+ "created_by": t.CreatedBy,
+ "created_at": t.CreatedAt,
+ })
+ }
+ h.WriteJSON(w, util.MapStr{"tokens": out}, http.StatusOK)
+}
+
+// createEnrollmentTokenHandler — POST /instance/_enrollment_tokens
+//
+// Body: {"name": "web-tier rollout", "max_uses": 50, "ttl_hours": 24}
+// Response carries the plaintext exactly once.
+func (h *APIHandler) createEnrollmentTokenHandler(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var body struct {
+ Name string `json:"name"`
+ MaxUses int `json:"max_uses"`
+ TTLHours int `json:"ttl_hours"`
+ }
+ if err := h.DecodeJSON(req, &body); err != nil {
+ h.WriteError(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if body.MaxUses <= 0 {
+ body.MaxUses = 1
+ }
+ if body.MaxUses > 10000 {
+ h.WriteError(w, "max_uses too large (max 10000)", http.StatusBadRequest)
+ return
+ }
+ ttl := time.Duration(body.TTLHours) * time.Hour
+ if body.TTLHours <= 0 {
+ ttl = 24 * time.Hour
+ }
+
+ rec, plaintext, err := mintEnrollmentToken(body.Name, body.MaxUses, ttl, req.Header.Get("X-API-USER"))
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ if err := orm.Save(ctx, rec); err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{
+ "id": rec.ID,
+ "name": rec.Name,
+ "token": plaintext, // shown exactly once
+ "max_uses": rec.MaxUses,
+ "expires_at": rec.ExpiresAt,
+ }, http.StatusOK)
+}
+
+// revokeEnrollmentTokenHandler — DELETE /instance/_enrollment_tokens/:id
+func (h *APIHandler) revokeEnrollmentTokenHandler(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+ id := ps.ByName("id")
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ rec := EnrollmentToken{}
+ rec.ID = id
+ exists, err := orm.GetV2(ctx, &rec)
+ if err != nil || !exists {
+ h.WriteOpRecordNotFoundJSON(w, id)
+ return
+ }
+ rec.Revoked = true
+ if err := orm.Save(ctx, &rec); err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{"id": id, "revoked": true}, http.StatusOK)
+}
diff --git a/modules/configs/server/enrollment_token_test.go b/modules/configs/server/enrollment_token_test.go
new file mode 100644
index 000000000..2a4206b4f
--- /dev/null
+++ b/modules/configs/server/enrollment_token_test.go
@@ -0,0 +1,72 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "testing"
+ "time"
+
+ httptest "net/http/httptest"
+)
+
+func TestRateLimiter(t *testing.T) {
+ rl := newRateLimiter(time.Minute, 3)
+ ip := "10.0.0.1"
+ for i := 0; i < 3; i++ {
+ if !rl.allow(ip) {
+ t.Fatalf("hit %d should be allowed (max 3)", i+1)
+ }
+ }
+ if rl.allow(ip) {
+ t.Fatal("4th hit in the same window must be denied")
+ }
+ // different key unaffected
+ if !rl.allow("10.0.0.2") {
+ t.Fatal("different IP must not share the bucket")
+ }
+ // window rollover
+ rl.hits[ip].since = time.Now().Add(-2 * time.Minute)
+ if !rl.allow(ip) {
+ t.Fatal("new window must reset the bucket")
+ }
+ // disabled limiter
+ var off *rateLimiter
+ if !off.allow(ip) {
+ t.Fatal("nil limiter = disabled = always allow")
+ }
+}
+
+func TestClientIP(t *testing.T) {
+ req := httptest.NewRequest("POST", "/x", nil)
+ req.RemoteAddr = "1.2.3.4:5678"
+ if got := clientIP(req); got != "1.2.3.4" {
+ t.Fatalf("remote addr parse = %q", got)
+ }
+ req.Header.Set("X-Forwarded-For", "9.9.9.9, 10.0.0.1")
+ if got := clientIP(req); got != "9.9.9.9" {
+ t.Fatalf("proxy chain = %q", got)
+ }
+}
+
+func TestEnrollmentTokenMint(t *testing.T) {
+ rec, plaintext, err := mintEnrollmentToken("rollout", 5, time.Hour, "admin")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(plaintext) <= len(enrollmentTokenPrefix) || plaintext[:len(enrollmentTokenPrefix)] != enrollmentTokenPrefix {
+ t.Fatalf("plaintext = %q, want %s-... prefix", plaintext, enrollmentTokenPrefix)
+ }
+ if rec.TokenHash == plaintext {
+ t.Fatal("hash must differ from plaintext")
+ }
+ if rec.MaxUses != 5 || rec.UsedCount != 0 {
+ t.Fatalf("policy = %d/%d", rec.UsedCount, rec.MaxUses)
+ }
+ // deterministic hash of the same plaintext
+ rec2, p2, _ := mintEnrollmentToken("x", 1, time.Hour, "")
+ if rec2.TokenHash == rec.TokenHash && plaintext != p2 {
+ t.Fatal("different plaintexts must not share a hash")
+ }
+}
diff --git a/modules/configs/server/instance_token.go b/modules/configs/server/instance_token.go
new file mode 100644
index 000000000..c82ec119f
--- /dev/null
+++ b/modules/configs/server/instance_token.go
@@ -0,0 +1,190 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "crypto/rand"
+ "crypto/sha256"
+ "crypto/subtle"
+ "encoding/hex"
+ "net/http"
+ "time"
+
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
+ "infini.sh/framework/core/model"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Per-instance manager tokens (console_design: managed-config-security.md §4).
+//
+// Lifecycle (portable adaptation of Console's managed token flow — the
+// final-state design from pr/framework-managed-token-flow-20260524 as
+// evolved on console_framework, simplified for the embedded server):
+//
+// 1. BOOTSTRAP: a fresh instance registers using one of the statically
+// configured tokens (configs.server.auth.tokens). The server mints an
+// instance-scoped token and returns it in the register response.
+// 2. STEADY STATE: the instance presents its per-instance token
+// (Authorization: Bearer) on every /configs/_sync.
+// 3. ROTATION: POST /instance/_exchange_token with the current token
+// mints a replacement; the previous token stays valid for 1h
+// (grace window for in-flight syncs).
+//
+// Storage: only the SHA-256 hash of each token is persisted (tokens are
+// bearer-equivalent secrets; hashing at rest means a database leak does not
+// leak credentials). Comparison is constant-time.
+// ──────────────────────────────────────────────────────────────────────────
+
+// rotationGrace is how long a superseded token remains valid after exchange.
+const rotationGrace = time.Hour
+
+// InstanceToken is the per-instance manager credential record.
+type InstanceToken struct {
+ orm.ORMObjectBase
+
+ InstanceID string `json:"instance_id" elastic_mapping:"instance_id:{type:keyword}"`
+ // TokenHash is sha256(token) of the current token.
+ TokenHash string `json:"token_hash" elastic_mapping:"token_hash:{type:keyword}"`
+ // PreviousHash is sha256(token) of the superseded token (rotation grace).
+ PreviousHash string `json:"previous_hash,omitempty" elastic_mapping:"previous_hash:{type:keyword}"`
+ // RotatedAt is when the current token was minted (grace window anchor).
+ RotatedAt time.Time `json:"rotated_at" elastic_mapping:"rotated_at:{type:date}"`
+}
+
+// MintInstanceToken creates (or rotates) the token record for an instance
+// and returns the plaintext token — the ONLY time it exists in the clear
+// outside the client's memory.
+func MintInstanceToken(ctx *orm.Context, instanceID string) (string, error) {
+ raw := make([]byte, 32)
+ if _, err := rand.Read(raw); err != nil {
+ return "", err
+ }
+ token := hex.EncodeToString(raw)
+
+ rec := loadInstanceToken(ctx, instanceID)
+ if rec == nil {
+ rec = &InstanceToken{InstanceID: instanceID}
+ rec.ID = util.GetUUID()
+ } else {
+ rec.PreviousHash = rec.TokenHash
+ }
+ rec.TokenHash = hashToken(token)
+ rec.RotatedAt = time.Now().UTC()
+ orm.WithModel(ctx, rec)
+ if err := orm.Save(ctx, rec); err != nil {
+ return "", err
+ }
+ return token, nil
+}
+
+// ValidateInstanceToken checks a presented token against the instance's
+// current record: current token always passes; the previous token passes
+// only inside the rotation grace window. Constant-time per comparison.
+func ValidateInstanceToken(ctx *orm.Context, instanceID, token string) bool {
+ if token == "" {
+ return false
+ }
+ rec := loadInstanceToken(ctx, instanceID)
+ if rec == nil {
+ return false
+ }
+ got := hashToken(token)
+ match := subtle.ConstantTimeCompare([]byte(got), []byte(rec.TokenHash))
+ if match == 1 {
+ return true
+ }
+ if rec.PreviousHash != "" && time.Since(rec.RotatedAt) < rotationGrace {
+ return subtle.ConstantTimeCompare([]byte(got), []byte(rec.PreviousHash)) == 1
+ }
+ return false
+}
+
+func loadInstanceToken(ctx *orm.Context, instanceID string) *InstanceToken {
+ orm.WithModel(ctx, &InstanceToken{})
+ qb := orm.NewQuery().
+ Filter(orm.TermQuery("instance_id", instanceID)).
+ Size(1)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil || res == nil {
+ return nil
+ }
+ tokens, _, _ := elastic.DecodeHits[InstanceToken](res)
+ if len(tokens) == 0 {
+ return nil
+ }
+ return &tokens[0]
+}
+
+func hashToken(token string) string {
+ sum := sha256.Sum256([]byte(token))
+ return hex.EncodeToString(sum[:])
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// HTTP: token exchange
+// ──────────────────────────────────────────────────────────────────────────
+
+// exchangeTokenHandler — POST /instance/_exchange_token
+//
+// Body: {"instance_id": "..."} authenticated by the CURRENT token (Bearer).
+// Response: {"manager_token": "", "grace_seconds": 3600}
+func (h *APIHandler) exchangeTokenHandler(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var body struct {
+ InstanceID string `json:"instance_id"`
+ }
+ if err := h.DecodeJSON(req, &body); err != nil || body.InstanceID == "" {
+ h.WriteError(w, "instance_id is required", http.StatusBadRequest)
+ return
+ }
+
+ presented := extractBearerToken(req)
+ if presented == "" {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+
+ // Accepted credentials: the current standard manager token, the
+ // instance's registered self API token (the exchange's whole purpose:
+ // self token → manager token), or a static admin token.
+ ok := matchesManagerToken(ctx, body.InstanceID, presented)
+ if !ok {
+ ok = matchesRegisteredAccessToken(ctx, body.InstanceID, presented)
+ }
+ if !ok {
+ ok = validateStaticToken(presented)
+ }
+ if !ok {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+
+ // Mint the framework-standard manager token (access_token machinery:
+ // stored in KV, validated by matchesManagerToken). The legacy
+ // InstanceToken scheme (MintInstanceToken) is NOT accepted by sync
+ // anymore — returning it here poisoned clients' credentials.
+ instanceName := ""
+ nameInst := model.Instance{}
+ nameInst.ID = body.InstanceID
+ if exists, err := orm.GetV2(ctx, &nameInst); err == nil && exists {
+ instanceName = nameInst.Name
+ }
+ token, err := mintManagerToken(body.InstanceID, instanceName)
+ if err != nil {
+ h.WriteError(w, "mint token: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{
+ "manager_token": token,
+ "manager_api_token": token, // agent-side key (managed.ExchangeTokens)
+ "grace_seconds": int(rotationGrace.Seconds()),
+ }, http.StatusOK)
+}
diff --git a/modules/configs/server/manager_token.go b/modules/configs/server/manager_token.go
new file mode 100644
index 000000000..0e5540f22
--- /dev/null
+++ b/modules/configs/server/manager_token.go
@@ -0,0 +1,126 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "time"
+
+ log "github.com/cihub/seelog"
+
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/security"
+ "infini.sh/framework/modules/security/access_token"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Manager tokens — the framework's standard access-token manager backs the
+// post-approval credential (replacing the earlier custom InstanceToken
+// scheme). Approving an instance mints a real AccessToken via
+// access_token.CreateAPIToken:
+//
+// - stored/validated/revoked through the standard machinery
+// (KV fast-lookup + ORM record, the token management UI/API)
+// - carries instance binding (Data.instance_id) so a token can never
+// authenticate a different instance
+// - permissions attachable later for scoped manager capabilities
+//
+// The plaintext is returned exactly once (register/approve response) —
+// the same one-time-display rule as before.
+// ──────────────────────────────────────────────────────────────────────────
+
+const managerTokenType = "managed_instance"
+
+// mintManagerToken creates a framework-standard access token bound to the
+// instance. Returns the plaintext.
+// maxManagerTokensPerInstance 每实例保留的 manager token 上限 (轮换容错:
+// 旧 token 在新 token 送达实例前仍需可用; 再旧的属泄漏, 铸造时修剪)。
+const maxManagerTokensPerInstance = 2
+
+// pruneManagerTokens 删除同实例超出上限的旧 manager token。
+func pruneManagerTokens(instanceID string) {
+ tokens, err := access_token.ListTokens()
+ if err != nil {
+ return
+ }
+ type entry struct {
+ id string
+ seq string
+ }
+ var mine []entry
+ for i := range tokens {
+ t := tokens[i] // 指针遍历: AccessToken 内含锁, 值拷贝被 vet 捕获
+ if t.Type != managerTokenType {
+ continue
+ }
+ if bound, _ := t.Data["instance_id"].(string); bound != instanceID {
+ continue
+ }
+ mine = append(mine, entry{id: t.ID, seq: t.ID})
+ }
+ if len(mine) <= maxManagerTokensPerInstance {
+ return
+ }
+ // id 是时序生成的 (k-sortid), 字典序即时间序: 保留最后 N 个。
+ for i := 0; i < len(mine)-maxManagerTokensPerInstance; i++ {
+ if err := access_token.DeleteTokenByID(mine[i].id); err == nil {
+ log.Infof("configs server: pruned stale manager token %s for instance %s", mine[i].id, instanceID)
+ }
+ }
+}
+
+func mintManagerToken(instanceID, instanceName string) (string, error) {
+ user := &security.UserSessionInfo{
+ Provider: "configs_server",
+ Login: instanceID,
+ }
+ user.SetUserID(instanceID)
+ user.Set("instance_id", instanceID)
+ if instanceName != "" {
+ user.Set("instance_name", instanceName)
+ }
+
+ res, err := access_token.CreateAPIToken(user,
+ "manager "+instanceName, "manager credential for instance "+instanceID,
+ managerTokenType, -1, nil)
+ if err != nil {
+ return "", err
+ }
+ token, _ := res["access_token"].(string)
+ if token != "" {
+ pruneManagerTokens(instanceID)
+ }
+ return token, nil
+}
+
+// matchesManagerToken reports whether the presented token is a valid
+// framework access token minted FOR THIS INSTANCE (manager binding).
+func matchesManagerToken(_ *orm.Context, instanceID, presented string) bool {
+ if presented == "" {
+ return false
+ }
+ t, err := access_token.GetToken(presented)
+ if err != nil || t == nil {
+ return false
+ }
+ if t.Type != managerTokenType {
+ return false
+ }
+ if t.ExpireIn > 0 && t.ExpireIn < time.Now().Unix() {
+ return false
+ }
+ bound, _ := t.Data["instance_id"].(string)
+ return bound == instanceID
+}
+
+// MintPublicManagerToken is mintManagerToken for admin surfaces (rotation).
+func MintPublicManagerToken(instanceID, instanceName string) (string, error) {
+ return mintManagerToken(instanceID, instanceName)
+}
+
+// EnrollmentRequired reports whether the admission ticket gate is on.
+func EnrollmentRequired() bool { return serverConfig.Enrollment.Required }
+
+// StaticTokens returns the configured static gate tokens (empty = open mode).
+func StaticTokens() []string { return staticTokens }
diff --git a/modules/configs/server/reverse_channel.go b/modules/configs/server/reverse_channel.go
new file mode 100644
index 000000000..a93821901
--- /dev/null
+++ b/modules/configs/server/reverse_channel.go
@@ -0,0 +1,154 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "fmt"
+ "net/http"
+ "strings"
+ "sync"
+
+ "infini.sh/framework/core/api"
+ framework_ws "infini.sh/framework/core/api/websocket"
+ "infini.sh/framework/core/api/websocket/reverse"
+ log "infini.sh/framework/core/log"
+ "infini.sh/framework/core/model"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+)
+
+// ──────────────────────────────────────────────────────────────────────────
+// Reverse channel hosting — managed instances are NOT directly reachable:
+// agents sit behind NAT/firewalls and only dial OUT. The manager calls
+// back THROUGH the agent-initiated websocket (see
+// core/api/websocket/reverse): the agent connects to the manager's /ws
+// endpoint with its instance ID in the peer header, HELLOs, and from
+// then on the manager can ProxyRequest down that connection while the
+// agent executes locally and streams the response back.
+//
+// This file wires the manager side: websocket callbacks + commands, and
+// a ReverseProxyRequest helper for consumers (LogPilot's instance
+// detail panel is the first).
+// ──────────────────────────────────────────────────────────────────────────
+
+var (
+ reverseManager *reverse.SessionManager
+ reverseRegisterOnce sync.Once
+)
+
+// ReverseChannelReady reports whether the agent reverse channel is wired.
+func ReverseChannelReady() bool { return reverseManager != nil }
+
+// ReverseIsConnected reports whether the given instance holds an active
+// reverse-channel session (i.e., the manager can reach it right now).
+func ReverseIsConnected(instanceID string) bool {
+ return reverseManager != nil && reverseManager.IsConnected(instanceID)
+}
+
+// ReverseProxyRequest performs a logical HTTP request against a managed
+// instance THROUGH its reverse channel. Returns an error when the
+// instance is not connected (the one-way deployment reality).
+func ReverseProxyRequest(peerID string, req *util.Request) (*util.Result, error) {
+ if reverseManager == nil {
+ return nil, fmt.Errorf("reverse channel not enabled on this manager")
+ }
+ // The send callback pushes the wire frame down the instance's
+ // websocket session.
+ send := func(sessionID, payload string) error {
+ return framework_ws.SendPrivateMessage(sessionID, payload)
+ }
+ return reverseManager.ProxyRequest(peerID, req, nil, send, nil)
+}
+
+// ReverseProxyRequestJSON is ReverseProxyRequest with a JSON response
+// unmarshal convenience.
+func ReverseProxyRequestJSON(peerID string, req *util.Request, out interface{}) (*util.Result, error) {
+ if reverseManager == nil {
+ return nil, fmt.Errorf("reverse channel not enabled on this manager")
+ }
+ send := func(sessionID, payload string) error {
+ return framework_ws.SendPrivateMessage(sessionID, payload)
+ }
+ return reverseManager.ProxyRequest(peerID, req, nil, send, out)
+}
+
+// setupReverseChannel wires the websocket callbacks and commands. Called
+// from Setup once.
+func setupReverseChannel() {
+ reverseRegisterOnce.Do(func() {
+ reverseManager = reverse.NewSessionManager(reverse.ManagerOptions{})
+
+ framework_ws.RegisterConnectCallback(onReverseConnect)
+ framework_ws.RegisterDisconnectCallback(onReverseDisconnect)
+ api.HandleWebSocketCommand(reverse.HelloCommand, "instance reverse hello", handleReverseHello)
+ api.HandleWebSocketCommand(reverse.ResponseCommand, "instance reverse response", handleReverseResponse)
+
+ log.Info("configs server: reverse channel ready (instances dial /ws; manager calls back through it)")
+ })
+}
+
+// onReverseConnect validates the connecting instance and opens its
+// pending session. The agent sends its instance ID in the peer header.
+func onReverseConnect(sessionID string, w http.ResponseWriter, r *http.Request) error {
+ instanceID := strings.TrimSpace(r.Header.Get(reverse.HeaderPeerID))
+ if instanceID == "" {
+ return nil // not a managed-instance connection
+ }
+
+ ctx := orm.NewContext().DirectAccess()
+ inst := model.Instance{}
+ inst.ID = instanceID
+ exists, err := orm.GetV2(ctx, &inst)
+ if err != nil && !isNotFound(err) {
+ return err
+ }
+ if err != nil || !exists {
+ return fmt.Errorf("instance %s is not registered", instanceID)
+ }
+ if loadInstanceStatus(ctx, instanceID) != StatusApproved {
+ return fmt.Errorf("instance %s is not approved", instanceID)
+ }
+ // Credential check: the dial-out must authenticate like a sync would
+ // (manager token / registered self token in the Authorization header
+ // or token query param).
+ presented := strings.TrimSpace(r.Header.Get("Authorization"))
+ presented = strings.TrimPrefix(presented, "Bearer ")
+ if presented == "" {
+ presented = strings.TrimSpace(r.URL.Query().Get("token"))
+ }
+ if !matchesManagerToken(ctx, instanceID, presented) &&
+ !matchesRegisteredAccessToken(ctx, instanceID, presented) {
+ return fmt.Errorf("instance %s reverse channel credential rejected", instanceID)
+ }
+
+ reverseManager.RegisterPendingSession(sessionID, instanceID)
+ return nil
+}
+
+func onReverseDisconnect(sessionID string) {
+ if reverseManager != nil {
+ reverseManager.OnDisconnect(sessionID)
+ }
+}
+
+func handleReverseHello(c *framework_ws.WebsocketConnection, array []string) {
+ if len(array) < 2 || reverseManager == nil {
+ return
+ }
+ if err := reverseManager.HandleHelloPayload(strings.Join(array[1:], " ")); err != nil {
+ log.Warnf("configs server: reverse hello rejected: %v", err)
+ } else {
+ log.Info("configs server: reverse hello accepted")
+ }
+}
+
+func handleReverseResponse(c *framework_ws.WebsocketConnection, array []string) {
+ if len(array) < 2 || reverseManager == nil {
+ return
+ }
+ if err := reverseManager.HandleResponsePayload(strings.Join(array[1:], " ")); err != nil {
+ log.Debugf("configs server: reverse response error: %v", err)
+ }
+}
diff --git a/modules/configs/server/server.go b/modules/configs/server/server.go
new file mode 100644
index 000000000..8e8e20457
--- /dev/null
+++ b/modules/configs/server/server.go
@@ -0,0 +1,803 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package server implements the SERVER side of the standard managed-config
+// protocol: the counterpart of modules/configs/client that every framework
+// process runs when `configs.managed: true`.
+//
+// Routes (protocol-compatible with Console's managed plugin; Console keeps
+// its richer implementation — token exchange, script hooks, websocket
+// proxy — on its own product surface, this one carries the portable core
+// any product can embed):
+//
+// POST /instance/_register self-description registration (upsert)
+// POST /configs/_sync heartbeat + config diff delivery
+//
+// Sync semantics (mirrors the Console contract):
+// - the client posts its current managed configs + a hash of them
+// - the server replies {changed, configs:{created,updated,deleted}}
+// - unchanged content is skipped by version comparison; a config the
+// client marked Managed=false is never touched
+// - every sync refreshes the instance record (heartbeat via labels)
+package server
+
+import (
+ "crypto/subtle"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "strconv"
+ "strings"
+ "time"
+
+ "infini.sh/framework/core/api"
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
+ "infini.sh/framework/core/env"
+ log "infini.sh/framework/core/log"
+ "infini.sh/framework/core/model"
+ "infini.sh/framework/core/orm"
+ "infini.sh/framework/core/util"
+ "infini.sh/framework/lib/go-ucfg"
+ "infini.sh/framework/modules/configs/common"
+)
+
+// Config holds the server-side settings (configs.server.* in YAML).
+// Disabled by default; products opt in via Setup().
+type Config struct {
+ Enabled bool `config:"enabled"`
+
+ // Auth gates both protocol routes (_register/_sync) with the standard
+ // Bearer token mechanism: clients present
+ // Authorization: Bearer (or X-API-Token: )
+ // and the server constant-time-compares against the configured token
+ // list. Multiple tokens allow zero-downtime rotation (add the new
+ // token, roll the clients, remove the old one). Deployments MUST
+ // configure auth.tokens in production: without it any host that
+ // reaches the port can register instances and pull every assigned
+ // config. When unset the server logs a warning and runs in open dev
+ // mode. The framework client sends its token automatically via
+ // configs.manager.token.
+ Auth struct {
+ Tokens []ucfg.SecretString `config:"tokens"`
+ } `config:"auth"`
+
+ // Enrollment requires a valid one-time admission ticket on register
+ // (see enrollment_token.go). false = open registration (admission
+ // flow still gates credentials/configs behind manual approval).
+ Enrollment struct {
+ Required bool `config:"required"`
+ } `config:"enrollment"`
+
+ // RegisterRateLimit caps registration attempts per client IP
+ // (fixed window). Defaults: 10 per minute; 0 disables.
+ RegisterRateLimit struct {
+ MaxHits int `config:"max_hits"`
+ Window string `config:"window"`
+ } `config:"register_rate_limit"`
+}
+
+// ManagedConfig is one config file assigned to instances — by explicit
+// instance id, by group membership (Groups), or "*" (all instances).
+// Version bumps on every content change; clients apply Created/Updated
+// diffs by version comparison.
+type ManagedConfig struct {
+ orm.ORMObjectBase
+
+ InstanceID string `json:"instance_id" elastic_mapping:"instance_id:{type:keyword}"` // target instance id, "*" for all, "" when Groups targeting is used
+ Groups []string `json:"groups,omitempty" elastic_mapping:"groups:{type:keyword}"` // target instance groups (any-match)
+ Name string `json:"name" elastic_mapping:"name:{type:keyword}"` // config file name, e.g. pipeline.yml
+ Location string `json:"location,omitempty" elastic_mapping:"location:{type:keyword}"`
+ Content string `json:"content,omitempty" elastic_mapping:"content:{type:text}"`
+ Version int64 `json:"version" elastic_mapping:"version:{type:long}"`
+ Readonly bool `json:"readonly,omitempty" elastic_mapping:"readonly:{type:boolean}"`
+}
+
+// Label keys for heartbeat state on the instance record (model.Instance
+// has no dedicated online-state fields; labels keep the wire type intact).
+const (
+ LabelLastSyncAt = "managed_last_sync_at"
+ LabelRegistered = "managed_registered"
+)
+
+// Instance admission states (Instance.Status).
+const (
+ StatusPending = "pending" // registered, awaiting admin approval
+ StatusApproved = "approved" // admitted; full sync + credentials
+)
+
+// AllInstancesID assigns a ManagedConfig to every syncing instance.
+const AllInstancesID = "*"
+
+// instanceTokenExchangeAPI rotates an instance's manager token.
+const instanceTokenExchangeAPI = "/instance/_exchange_token"
+
+// instanceApproveAPI admits a pending instance (management action).
+const instanceApproveAPI = "/instance/:id/_approve"
+
+// enrollmentTokensAPI manages one-time registration tickets.
+const enrollmentTokensAPI = "/instance/_enrollment_tokens"
+
+type APIHandler struct {
+ api.Handler
+}
+
+var handler = &APIHandler{}
+
+// registerLimiter rate-limits the publicly reachable register endpoint.
+var registerLimiter *rateLimiter
+
+// serverConfig is the parsed configs.server section (set in Setup).
+var serverConfig Config
+
+// Setup registers the ORM schemas and the protocol routes. Call once from
+// the product's module setup (e.g. logpilot's init). No-op when
+// configs.server.enabled is false in the product config.
+func Setup() {
+ cfg := Config{Enabled: true}
+ exists, err := env.ParseConfig("configs.server", &cfg)
+ if err != nil {
+ panic(err)
+ }
+ if exists && !cfg.Enabled {
+ log.Debug("configs server disabled by configuration")
+ return
+ }
+
+ setupReverseChannel()
+
+ orm.MustRegisterSchemaWithIndexName(model.Instance{}, "instance")
+ orm.MustRegisterSchemaWithIndexName(ManagedConfig{}, "managed-configs")
+ orm.MustRegisterSchemaWithIndexName(InstanceToken{}, "instance-tokens")
+ orm.MustRegisterSchemaWithIndexName(EnrollmentToken{}, enrollmentTokenModel)
+
+ window, werr := time.ParseDuration(cfg.RegisterRateLimit.Window)
+ if werr != nil || window <= 0 {
+ window = time.Minute
+ }
+ maxHits := cfg.RegisterRateLimit.MaxHits
+ if maxHits == 0 {
+ maxHits = 10
+ }
+ registerLimiter = newRateLimiter(window, maxHits)
+
+ serverConfig = cfg
+ gate := newTokenGate(cfg.Auth.Tokens)
+ registerGate := gate
+ if len(cfg.Auth.Tokens) == 0 {
+ // Open mode is warned about at startup; the per-instance flow below
+ // still mints tokens so closing the server later does not strand
+ // instances that already registered.
+ registerGate = func(next httprouter.Handle) httprouter.Handle { return next }
+ }
+ api.HandleAPIMethod(api.POST, common.REGISTER_API, registerGate(handler.registerInstance))
+ api.HandleAPIMethod(api.POST, common.SYNC_API, gate(handler.syncConfigs))
+ api.HandleAPIMethod(api.POST, instanceTokenExchangeAPI, gate(handler.exchangeTokenHandler))
+ // Admission: management UI approves pending instances. The gate admits
+ // admins (static token); the handler mints the instance credential.
+ api.HandleAPIMethod(api.POST, instanceApproveAPI, gate(handler.approveInstanceHandler))
+
+ api.HandleUIMethod(api.POST, common.REGISTER_API, registerGate(handler.registerInstance))
+ api.HandleUIMethod(api.POST, common.SYNC_API, gate(handler.syncConfigs))
+ api.HandleUIMethod(api.POST, instanceTokenExchangeAPI, gate(handler.exchangeTokenHandler))
+ api.HandleUIMethod(api.POST, instanceApproveAPI, gate(handler.approveInstanceHandler))
+
+ // Enrollment-token management (admin, token-gated).
+ api.HandleAPIMethod(api.GET, enrollmentTokensAPI, gate(handler.enrollmentTokensHandler))
+ api.HandleAPIMethod(api.POST, enrollmentTokensAPI, gate(handler.createEnrollmentTokenHandler))
+ api.HandleAPIMethod(api.DELETE, enrollmentTokensAPI+"/:id", gate(handler.revokeEnrollmentTokenHandler))
+ api.HandleUIMethod(api.GET, enrollmentTokensAPI, gate(handler.enrollmentTokensHandler))
+ api.HandleUIMethod(api.POST, enrollmentTokensAPI, gate(handler.createEnrollmentTokenHandler))
+ api.HandleUIMethod(api.DELETE, enrollmentTokensAPI+"/:id", gate(handler.revokeEnrollmentTokenHandler))
+
+ if len(cfg.Auth.Tokens) > 0 {
+ log.Infof("configs server ready: %s + %s (bearer token auth, %d token(s) accepted)", common.REGISTER_API, common.SYNC_API, len(cfg.Auth.Tokens))
+ } else {
+ log.Warnf("configs server ready in OPEN mode (no configs.server.auth.tokens configured) - " +
+ "any host reaching this port can register instances and pull assigned configs; configure auth.tokens in production")
+ }
+}
+
+// staticTokens holds the configured bootstrap/admin tokens (set in Setup).
+var staticTokens []string
+
+// validateStaticToken constant-time checks against the configured static
+// tokens (bootstrap admission + admin fallback).
+func validateStaticToken(token string) bool {
+ if token == "" || len(staticTokens) == 0 {
+ return false
+ }
+ matched := 0
+ for _, want := range staticTokens {
+ matched |= subtle.ConstantTimeCompare([]byte(token), []byte(want))
+ }
+ return matched == 1
+}
+
+// extractBearerToken reads the access token from the standard
+// Authorization: Bearer header, falling back to X-API-Token (the framework's
+// conventional token header).
+func extractBearerToken(req *http.Request) string {
+ if h := req.Header.Get("Authorization"); strings.HasPrefix(h, "Bearer ") {
+ return strings.TrimSpace(h[len("Bearer "):])
+ }
+ return strings.TrimSpace(req.Header.Get("X-API-Token"))
+}
+
+// newTokenGate wraps a protocol handler with constant-time Bearer-token
+// validation against the accepted token list. No tokens configured = open
+// pass-through (dev mode, loudly warned at startup).
+func newTokenGate(tokens []ucfg.SecretString) func(httprouter.Handle) httprouter.Handle {
+ if len(tokens) == 0 {
+ return func(next httprouter.Handle) httprouter.Handle { return next }
+ }
+ wants := make([]string, 0, len(tokens))
+ for _, t := range tokens {
+ if v := t.Get(); v != "" {
+ wants = append(wants, v)
+ }
+ }
+ staticTokens = wants
+ return func(next httprouter.Handle) httprouter.Handle {
+ return func(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+ got := extractBearerToken(req)
+ if got == "" {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ // Constant-time compare against EVERY accepted token (all of
+ // them, so the response time does not reveal which position
+ // matched); any match passes.
+ matched := 0
+ for _, want := range wants {
+ matched |= subtle.ConstantTimeCompare([]byte(got), []byte(want))
+ }
+ if matched != 1 {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ next(w, req, ps)
+ }
+ }
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// POST /instance/_register
+// ──────────────────────────────────────────────────────────────────────────
+
+// registerBody decodes both the wrapped form ({client:{...}}) and the
+// legacy plain model.Instance the framework client sends.
+type registerBody struct {
+ Client model.Instance `json:"client"`
+}
+
+func (h *APIHandler) registerInstance(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ // Flood control: the register endpoint is reachable by design; without
+ // a valid enrollment ticket this is the only guard against spam.
+ if !registerLimiter.allow(clientIP(req)) {
+ h.WriteError(w, "too many registration attempts", http.StatusTooManyRequests)
+ return
+ }
+
+ // Read the body ONCE: DecodeJSON consumes (and closes) r.Body, so a
+ // legacy-payload fallback that re-reads it fails with
+ // "invalid Read on closed Body" — which silently broke every bare
+ // model.Instance registration (the format deployed agents send).
+ body, err := readBody(req)
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ var instance model.Instance
+ if wrapped := struct {
+ Client model.Instance `json:"client"`
+ AccessToken *common.RegisterToken `json:"access_token"`
+ }{}; util.FromJSONBytes(body, &wrapped) == nil && wrapped.Client.ID != "" {
+ instance = wrapped.Client
+ // The framework client sends the agent's self API token at the
+ // WRAPPER level (common.InstanceRegisterRequest), not inside
+ // client — merge it so sync/reverse credential checks can match
+ // it (instance.AccessToken in the DB).
+ if wrapped.AccessToken != nil && strings.TrimSpace(wrapped.AccessToken.Value) != "" {
+ instance.AccessToken = &model.Token{Value: strings.TrimSpace(wrapped.AccessToken.Value)}
+ }
+ } else if err := util.FromJSONBytes(body, &instance); err != nil || instance.ID == "" {
+ h.WriteError(w, "instance id is required (plain Instance or {client:{...}} payload)", http.StatusBadRequest)
+ return
+ }
+ // Enrollment ticket applies to NEW registrations only. Re-registration
+ // needs no ticket when the instance already exists (admission already
+ // happened — the ticket got it in the door) or presents a persistent
+ // credential. Without the exists-check, an approved instance whose
+ // credential was never delivered (approve mints the token AFTER the
+ // one-use ticket was consumed) could never bootstrap: re-register 403
+ // forever. The re-register "exists with credential" check below still
+ // enforces identity for credentialed instances.
+ enrollmentCtx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ var validTicket *EnrollmentToken
+ presentedCred := extractBearerToken(req)
+ if serverConfig.Enrollment.Required {
+ instanceExists := false
+ probe := model.Instance{}
+ probe.ID = instance.ID
+ if exists, err := orm.GetV2(enrollmentCtx, &probe); err == nil && exists {
+ instanceExists = true
+ }
+ if instanceExists ||
+ matchesManagerToken(enrollmentCtx, instance.ID, presentedCred) ||
+ matchesRegisteredAccessToken(enrollmentCtx, instance.ID, presentedCred) {
+ log.Debugf("configs server: instance %s re-registers without an enrollment ticket (exists=%v)", instance.ID, instanceExists)
+ } else {
+ ticket := strings.TrimSpace(req.Header.Get("X-Enrollment-Token"))
+ if ticket == "" {
+ // also accept it in the payload wrapper for convenience
+ var probe struct {
+ EnrollmentToken string `json:"enrollment_token"`
+ }
+ _ = util.FromJSONBytes(body, &probe)
+ ticket = strings.TrimSpace(probe.EnrollmentToken)
+ }
+ validTicket = validateEnrollmentToken(enrollmentCtx, ticket)
+ if validTicket == nil {
+ log.Warnf("configs server: registration rejected for instance %s (invalid/expired/exhausted enrollment ticket)", instance.ID)
+ h.WriteError(w, "invalid enrollment token", http.StatusForbidden)
+ return
+ }
+ }
+ }
+
+ // The managed agent's self-generated API token rides in access_token
+ // (framework token management: the agent mints it via
+ // access_token.CreateAPIToken at startup and registers it here; the
+ // manager stores it for reverse calls — pipeline tasks, stats, proxy).
+ if instance.AccessToken != nil && strings.TrimSpace(instance.AccessToken.Value) != "" {
+ log.Debugf("configs server: instance %s registered an API access token", instance.ID)
+ }
+
+ // An EXISTING instance re-registering must prove identity with its own
+ // token (a static token also qualifies — bootstrap admin). A fresh
+ // instance is authenticated by the static gate already.
+ ormCtx := orm.NewContextWithParent(req.Context()).DirectAccess()
+ hasCredential := loadInstanceToken(ormCtx, instance.ID) != nil || instance.AccessToken != nil
+ if hasCredential {
+ ok := matchesManagerToken(ormCtx, instance.ID, presentedCred) ||
+ matchesRegisteredAccessToken(ormCtx, instance.ID, presentedCred) ||
+ validateStaticToken(presentedCred)
+ if !ok {
+ // Credential-rotation recovery: an APPROVED instance whose
+ // stored self token no longer matches (client-side state was
+ // wiped / reinstalled) may re-register — the upsert below
+ // rotates the stored self token to the presented one. Without
+ // this, such an instance is permanently locked out (its
+ // one-use enrollment ticket is long consumed).
+ if loadInstanceStatus(ormCtx, instance.ID) == StatusApproved && instance.AccessToken != nil {
+ ok = true
+ log.Infof("configs server: rotating self token for approved instance %s on re-register", instance.ID)
+ }
+ }
+ if !ok {
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized: instance credential required to re-register", http.StatusUnauthorized)
+ return
+ }
+ }
+
+ created, err := upsertInstance(&instance)
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+
+ // Admission: a PENDING instance is visible in the management UI but
+ // receives no credentials and no configs until an admin approves it.
+ // Admission state is SERVER-OWNED: read it back from the stored
+ // record (upsert preserves it), never from the incoming payload —
+ // agents don't send a status, so the payload would always read as
+ // pending and an approved agent would loop waiting for approval.
+ approved := loadInstanceStatus(ormCtx, instance.ID) == StatusApproved
+
+ if validTicket != nil && created {
+ // Burn a use only when this registration created the instance;
+ // re-registrations (pending agents retry every cycle) are free.
+ consumeEnrollmentToken(enrollmentCtx, validTicket)
+ }
+
+ resp := util.MapStr{
+ "id": instance.ID,
+ "approved": approved,
+ }
+ if approved {
+ // Mint the framework-standard manager token (AccessToken via
+ // access_token.CreateAPIToken — standard storage, revocation and
+ // UI); the response is the only place the plaintext ever appears.
+ token, err := mintManagerToken(instance.ID, instance.Name)
+ if err != nil {
+ h.WriteError(w, "mint manager token: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ resp["manager_token"] = token
+ }
+ if !created {
+ resp["exists"] = true // the framework client treats "exists" as success
+ } else {
+ resp["created"] = true
+ }
+ h.WriteJSON(w, resp, http.StatusOK)
+}
+
+// upsertInstance persists/refreshes the registration. Returns true when
+// the instance was newly created.
+func upsertInstance(instance *model.Instance) (bool, error) {
+ ctx := orm.NewContext().DirectAccess()
+ existing := model.Instance{}
+ existing.ID = instance.ID
+ exists, err := orm.GetV2(ctx, &existing)
+ if err != nil && !isNotFound(err) {
+ return false, err
+ }
+ // not-found is the normal first-registration path, not an error
+ exists = err == nil && exists
+
+ now := strconv.FormatInt(time.Now().UnixMilli(), 10)
+ if instance.Labels == nil {
+ instance.Labels = map[string]string{}
+ }
+ instance.Labels[LabelRegistered] = now
+ instance.Labels[LabelLastSyncAt] = now
+ if instance.Status == "" {
+ instance.Status = StatusPending
+ }
+
+ if exists {
+ // Keep server-owned fields; refresh the self-description only.
+ // Status is server-owned (admission) — the incoming payload has
+ // no say, otherwise every heartbeat/re-register would clobber an
+ // approved instance back to pending.
+ created := existing.Created
+ status := existing.Status
+ instanceCopy := *instance
+ instanceCopy.Created = created
+ instanceCopy.Status = status
+ // Groups are server-owned (UI-managed targeting): the heartbeat
+ // payload does not carry them — keep the stored value.
+ instanceCopy.Groups = existing.Groups
+ // Heartbeat syncs carry no access token; keep the registered one
+ // (wiping it here would 401 the very next sync).
+ if instanceCopy.AccessToken == nil {
+ instanceCopy.AccessToken = existing.AccessToken
+ }
+ return false, orm.Save(ctx, &instanceCopy)
+ }
+ created := time.Now().UTC()
+ instance.Created = &created
+ return true, orm.Save(ctx, instance)
+}
+
+// ──────────────────────────────────────────────────────────────────────────
+// POST /configs/_sync
+// ──────────────────────────────────────────────────────────────────────────
+
+func (h *APIHandler) syncConfigs(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ var obj common.ConfigSyncRequest
+ if err := h.DecodeJSON(req, &obj); err != nil {
+ h.WriteError(w, err.Error(), http.StatusBadRequest)
+ return
+ }
+ if obj.Client.ID == "" {
+ h.WriteError(w, "client.id is required", http.StatusBadRequest)
+ return
+ }
+
+ // Authentication: the static gate has already accepted the caller, but
+ // instances that hold per-instance tokens must be checked against them —
+ // a revoked static token must not keep a registered instance alive, and
+ // conversely a valid instance token must pass even if statics rotate.
+ // Accepted credentials: the minted InstanceToken (Bearer, from the
+ // register/exchange response) OR the agent's registered self API token
+ // (X-API-Token — what the framework client sends before exchange).
+ if loadInstanceStatus(orm.NewContext().DirectAccess(), obj.Client.ID) == StatusApproved {
+ // Credential enforcement applies to admitted instances only:
+ // pending ones hold no paired credential yet and their sync
+ // carries nothing sensitive (empty config set) — heartbeat
+ // visibility is exactly what pending needs.
+ // Accepted credentials: the framework-standard manager token
+ // (minted at approve/register through access_token.CreateAPIToken)
+ // or the agent's registered self API token (pre-exchange).
+ ormAuthCtx := orm.NewContext().DirectAccess()
+ presented := extractBearerToken(req)
+ mgrOK := matchesManagerToken(ormAuthCtx, obj.Client.ID, presented)
+ regOK := matchesRegisteredAccessToken(ormAuthCtx, obj.Client.ID, presented)
+ if !mgrOK && !regOK {
+ log.Warnf("configs server: sync rejected for %s (presented=%dB pfx=%s mgr=%v reg=%v)", obj.Client.ID, len(presented), safeTokenPrefix(presented), mgrOK, regOK)
+ w.Header().Set("WWW-Authenticate", `Bearer realm="configs"`)
+ h.WriteError(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ }
+
+ // Heartbeat: refresh the instance record on every sync.
+ if _, err := upsertInstance(&obj.Client); err != nil {
+ log.Debugf("configs server: heartbeat upsert failed for %s: %v", obj.Client.ID, err)
+ }
+
+ assigned, aerr := loadAssignedConfigs(obj.Client.ID)
+ if aerr != nil {
+ // Transient backend failure — defer to the next sync instead of
+ // diffing against an empty set (which would report every managed
+ // client file as Deleted).
+ log.Warnf("configs server: %v (instance %s sync deferred)", aerr, obj.Client.ID)
+ h.WriteError(w, "assigned configs temporarily unavailable", http.StatusServiceUnavailable)
+ return
+ }
+ if status := loadInstanceStatus(orm.NewContext().DirectAccess(), obj.Client.ID); status != StatusApproved { // empty (legacy) counts as pending
+ // Not approved yet (or status unknown): heartbeat counts, configs
+ // do not flow. The client keeps re-registering and will pick up
+ // approval on the next register/sync cycle.
+ assigned = nil
+ // Do NOT diff against the (empty) assigned set: a pending instance
+ // that already holds local managed files (e.g. it re-registered
+ // under a new identity but kept its data dir, or approval is
+ // momentarily missing) would otherwise have every local config
+ // reported as Deleted and wiped — breaking its running pipelines
+ // until an admin approves and republishes. Report "no change";
+ // the diff resumes once the instance is approved.
+ h.WriteJSON(w, common.ConfigSyncResponse{Changed: false}, http.StatusOK)
+ return
+ }
+
+ // Fast path: identical hash and no forced sync → nothing changed.
+ serverHash := ConfigsHash(assigned)
+ if !obj.ForceSync && obj.Hash != "" && obj.Hash == serverHash {
+ h.WriteJSON(w, common.ConfigSyncResponse{Changed: false}, http.StatusOK)
+ return
+ }
+
+ resp := diffConfigs(assigned, obj.Configs.Configs)
+ h.WriteJSON(w, resp, http.StatusOK)
+}
+
+// loadAssignedConfigs returns the server-side config files assigned to the
+// instance: its own + the "*" catch-all + any group-targeted config whose
+// Groups intersect the instance's groups. Newest version per name.
+// A query FAILURE returns an error — callers must NOT treat it as "no
+// configs": a transiently empty read (e.g. manager just booting, ORM not
+// ready) would diff every client file as Deleted and wipe managed configs
+// fleet-wide.
+func loadAssignedConfigs(instanceID string) ([]common.ConfigFile, error) {
+ ctx := orm.NewContext().DirectAccess()
+ orm.WithModel(ctx, &ManagedConfig{})
+
+ // OR semantics: per-instance rows + the "*" catch-all. (Repeated
+ // Filter() calls AND together — instance_id==id AND instance_id=="*"
+ // would match nothing.)
+ qb := orm.NewQuery().
+ Filter(orm.ShouldQuery(
+ orm.TermQuery("instance_id", instanceID),
+ orm.TermQuery("instance_id", AllInstancesID),
+ )).
+ Size(1000)
+ res, err := orm.SearchV2(ctx, qb)
+ if err != nil {
+ return nil, fmt.Errorf("assigned-config query failed: %w", err)
+ }
+ if res == nil {
+ return nil, fmt.Errorf("assigned-config query returned no result")
+ }
+ stored, _, _ := decodeManagedConfigs(res)
+
+ // Group-targeted configs: instance groups are server-owned, read from
+ // the record (never from the heartbeat payload).
+ if groups := loadInstanceGroups(ctx, instanceID); len(groups) > 0 {
+ groupSet := map[string]struct{}{}
+ for _, g := range groups {
+ groupSet[g] = struct{}{}
+ }
+ gres, gerr := orm.SearchV2(ctx, orm.NewQuery().
+ Filter(orm.ExistsQuery("groups")).
+ Size(1000))
+ if gerr != nil {
+ return nil, fmt.Errorf("group-config query failed: %w", gerr)
+ }
+ if gres != nil {
+ groupStored, _, _ := decodeManagedConfigs(gres)
+ for _, mc := range groupStored {
+ if !configMatchesAnyGroup(&mc, groupSet) {
+ continue
+ }
+ stored = append(stored, mc)
+ }
+ }
+ }
+
+ out := make([]common.ConfigFile, 0, len(stored))
+ for _, mc := range stored {
+ out = append(out, common.ConfigFile{
+ Name: mc.Name,
+ Location: mc.Location,
+ Content: mc.Content,
+ Version: mc.Version,
+ Managed: true,
+ Hash: util.MD5digest(mc.Content),
+ Size: int64(len(mc.Content)),
+ Updated: time.Now().UnixMilli(),
+ })
+ }
+ return out, nil
+}
+
+// configMatchesAnyGroup reports whether the config's target groups intersect
+// the instance's group set.
+func configMatchesAnyGroup(mc *ManagedConfig, groupSet map[string]struct{}) bool {
+ for _, g := range mc.Groups {
+ if _, ok := groupSet[strings.TrimSpace(g)]; ok {
+ return true
+ }
+ }
+ return false
+}
+
+// loadInstanceGroups returns the server-owned groups of an instance.
+func loadInstanceGroups(ctx *orm.Context, instanceID string) []string {
+ inst := model.Instance{}
+ inst.ID = instanceID
+ exists, err := orm.GetV2(ctx, &inst)
+ if err != nil || !exists {
+ return nil
+ }
+ return inst.Groups
+}
+
+// decodeManagedConfigs decodes search hits via the shared elastic mapper.
+func decodeManagedConfigs(res *orm.SearchResult) ([]ManagedConfig, int64, error) {
+ return elastic.DecodeHits[ManagedConfig](res)
+}
+
+// diffConfigs builds the protocol response: created (server-only),
+// updated (version newer than the client's), deleted (client-only).
+// Configs the client marked Managed=false are never touched.
+func diffConfigs(assigned []common.ConfigFile, clientConfigs map[string]common.ConfigFile) common.ConfigSyncResponse {
+ resp := common.ConfigSyncResponse{}
+ resp.Configs.CreatedConfigs = map[string]common.ConfigFile{}
+ resp.Configs.UpdatedConfigs = map[string]common.ConfigFile{}
+ resp.Configs.DeletedConfigs = map[string]common.ConfigFile{}
+
+ serverMap := map[string]common.ConfigFile{}
+ for _, c := range assigned {
+ serverMap[c.Name] = c
+ }
+
+ for name, sc := range serverMap {
+ cc, ok := clientConfigs[name]
+ if !ok {
+ resp.Configs.CreatedConfigs[name] = sc
+ continue
+ }
+ if !cc.Managed {
+ continue // client opted this file out of management
+ }
+ if sc.Version > cc.Version {
+ resp.Configs.UpdatedConfigs[name] = sc
+ }
+ }
+ for name, cc := range clientConfigs {
+ if _, ok := serverMap[name]; !ok {
+ if !cc.Managed {
+ continue
+ }
+ resp.Configs.DeletedConfigs[name] = cc
+ }
+ }
+
+ resp.Changed = len(resp.Configs.CreatedConfigs) > 0 ||
+ len(resp.Configs.UpdatedConfigs) > 0 ||
+ len(resp.Configs.DeletedConfigs) > 0
+ return resp
+}
+
+// ConfigsHash mirrors the framework client's hash: MD5 of the JSON of the
+// assigned config list, so both sides compare identical digests.
+func ConfigsHash(files []common.ConfigFile) string {
+ if len(files) == 0 {
+ return ""
+ }
+ b, err := json.Marshal(files)
+ if err != nil {
+ return ""
+ }
+ return util.MD5digest(string(b))
+}
+
+func readBody(req *http.Request) ([]byte, error) {
+ defer func() { _ = req.Body.Close() }()
+ return io.ReadAll(req.Body)
+}
+
+// isNotFound reports whether err is a backend not-found marker (sqlite's
+// ErrNotFound / elastic's ErrNotFound), which upsert treats as "create".
+func isNotFound(err error) bool {
+ return err != nil && strings.Contains(err.Error(), "not found")
+}
+
+// matchesRegisteredAccessToken constant-time compares the presented token
+// with the instance's registered self API token (Instance.AccessToken).
+func matchesRegisteredAccessToken(ctx *orm.Context, instanceID, presented string) bool {
+ if presented == "" {
+ return false
+ }
+ inst := model.Instance{}
+ inst.ID = instanceID
+ exists, err := orm.GetV2(ctx, &inst)
+ if err != nil || !exists || inst.AccessToken == nil {
+ return false
+ }
+ want := strings.TrimSpace(inst.AccessToken.Value)
+ if want == "" {
+ return false
+ }
+ return subtle.ConstantTimeCompare([]byte(presented), []byte(want)) == 1
+}
+
+// loadInstanceStatus returns the admission status of an instance ("" when
+// unknown).
+func loadInstanceStatus(ctx *orm.Context, instanceID string) string {
+ inst := model.Instance{}
+ inst.ID = instanceID
+ exists, err := orm.GetV2(ctx, &inst)
+ if err != nil || !exists {
+ return ""
+ }
+ return inst.Status
+}
+
+// approveInstanceHandler — POST /instance/:id/_approve
+//
+// Management action: flip a pending instance to approved and mint its
+// per-instance token. The instance receives the token on its next
+// register (the framework client re-registers while unapproved) or
+// exchange call.
+func (h *APIHandler) approveInstanceHandler(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+ id := ps.ByName("id")
+ ctx := orm.NewContextWithParent(req.Context()).DirectAccess()
+
+ inst := model.Instance{}
+ inst.ID = id
+ exists, err := orm.GetV2(ctx, &inst)
+ if err != nil && !isNotFound(err) {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ if err != nil || !exists {
+ h.WriteOpRecordNotFoundJSON(w, id)
+ return
+ }
+
+ if inst.Status != StatusApproved {
+ inst.Status = StatusApproved
+ if err := orm.Save(ctx, &inst); err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ log.Infof("configs server: instance %s (%s) approved", id, inst.Name)
+ }
+
+ token, err := mintManagerToken(id, inst.Name)
+ if err != nil {
+ h.WriteError(w, "mint manager token: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{"id": id, "status": StatusApproved, "manager_token": token}, http.StatusOK)
+}
+
+// safeTokenPrefix returns the first 8 chars of a token for log correlation
+// (never enough to brute-force, enough to identify which credential a
+// client is presenting).
+func safeTokenPrefix(token string) string {
+ if len(token) <= 8 {
+ return "****"
+ }
+ return token[:8]
+}
diff --git a/modules/configs/server/server_test.go b/modules/configs/server/server_test.go
new file mode 100644
index 000000000..65a138641
--- /dev/null
+++ b/modules/configs/server/server_test.go
@@ -0,0 +1,227 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package server
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/lib/go-ucfg"
+ "infini.sh/framework/modules/configs/common"
+)
+
+func cfg(name string, version int64) common.ConfigFile {
+ return common.ConfigFile{Name: name, Content: "content-of-" + name, Version: version, Managed: true}
+}
+
+func TestDiffConfigs_AllStates(t *testing.T) {
+ assigned := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 2)}
+
+ t.Run("fresh client gets everything created", func(t *testing.T) {
+ resp := diffConfigs(assigned, nil)
+ if !resp.Changed {
+ t.Fatal("expected changed=true")
+ }
+ if len(resp.Configs.CreatedConfigs) != 2 {
+ t.Fatalf("created = %d, want 2", len(resp.Configs.CreatedConfigs))
+ }
+ if len(resp.Configs.UpdatedConfigs) != 0 || len(resp.Configs.DeletedConfigs) != 0 {
+ t.Fatal("no updates/deletes expected")
+ }
+ })
+
+ t.Run("same versions → no change", func(t *testing.T) {
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1), "b.yml": cfg("b.yml", 2)}
+ resp := diffConfigs(assigned, client)
+ if resp.Changed {
+ t.Fatalf("expected no change, got %+v", resp.Configs)
+ }
+ })
+
+ t.Run("server version bump → updated", func(t *testing.T) {
+ bumped := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 3)}
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1), "b.yml": cfg("b.yml", 2)}
+ resp := diffConfigs(bumped, client)
+ if !resp.Changed {
+ t.Fatal("expected changed")
+ }
+ if len(resp.Configs.UpdatedConfigs) != 1 || resp.Configs.UpdatedConfigs["b.yml"].Version != 3 {
+ t.Fatalf("updated = %+v", resp.Configs.UpdatedConfigs)
+ }
+ })
+
+ t.Run("client-only config → deleted", func(t *testing.T) {
+ client := map[string]common.ConfigFile{
+ "a.yml": cfg("a.yml", 1),
+ "b.yml": cfg("b.yml", 2),
+ "gone.yml": cfg("gone.yml", 1),
+ }
+ resp := diffConfigs(assigned, client)
+ if !resp.Changed {
+ t.Fatal("expected changed")
+ }
+ if len(resp.Configs.DeletedConfigs) != 1 || resp.Configs.DeletedConfigs["gone.yml"].Name != "gone.yml" {
+ t.Fatalf("deleted = %+v", resp.Configs.DeletedConfigs)
+ }
+ })
+
+ t.Run("client opts out via Managed=false → untouched", func(t *testing.T) {
+ localOnly := cfg("local.yml", 1)
+ localOnly.Managed = false
+ client := map[string]common.ConfigFile{
+ "a.yml": cfg("a.yml", 1),
+ "b.yml": cfg("b.yml", 2),
+ "local.yml": localOnly,
+ }
+ resp := diffConfigs(assigned, client)
+ if resp.Changed {
+ t.Fatal("unmanaged local config must not trigger deletion")
+ }
+
+ // server-side version bump on a config the client holds unmanaged:
+ // also skipped
+ unmanagedA := cfg("a.yml", 1)
+ unmanagedA.Managed = false
+ client2 := map[string]common.ConfigFile{"a.yml": unmanagedA, "b.yml": cfg("b.yml", 2)}
+ resp2 := diffConfigs(assigned, client2)
+ if _, touched := resp2.Configs.UpdatedConfigs["a.yml"]; touched {
+ t.Fatal("unmanaged config must not be updated")
+ }
+ })
+
+ t.Run("server removed everything → all client configs deleted", func(t *testing.T) {
+ client := map[string]common.ConfigFile{"a.yml": cfg("a.yml", 1)}
+ resp := diffConfigs(nil, client)
+ if !resp.Changed || len(resp.Configs.DeletedConfigs) != 1 {
+ t.Fatalf("expected deletion, got %+v", resp.Configs)
+ }
+ })
+
+ t.Run("empty both sides → no change", func(t *testing.T) {
+ resp := diffConfigs(nil, nil)
+ if resp.Changed {
+ t.Fatal("expected changed=false")
+ }
+ })
+}
+
+func TestConfigsHash(t *testing.T) {
+ files := []common.ConfigFile{cfg("a.yml", 1), cfg("b.yml", 2)}
+ h1 := ConfigsHash(files)
+ h2 := ConfigsHash(files)
+ if h1 == "" {
+ t.Fatal("hash must not be empty")
+ }
+ if h1 != h2 {
+ t.Fatal("hash must be stable")
+ }
+ // 顺序无关不应成立? 协议要求两侧列表一致 — 同序序列化, 顺序变化视为变更
+ reordered := []common.ConfigFile{files[1], files[0]}
+ if ConfigsHash(reordered) == h1 {
+ t.Log("note: hash is order-sensitive (both sides marshal the same list)")
+ }
+ if ConfigsHash(nil) != "" {
+ t.Fatal("empty list must hash to empty string")
+ }
+ if ConfigsHash([]common.ConfigFile{cfg("a.yml", 2)}) == h1 {
+ t.Fatal("version change must change the hash")
+ }
+}
+
+func TestTokenGate(t *testing.T) {
+ handler := func(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ w.WriteHeader(http.StatusOK)
+ }
+ mkTokens := func(vals ...string) []ucfg.SecretString {
+ out := make([]ucfg.SecretString, len(vals))
+ for i, v := range vals {
+ out[i] = ucfg.SecretString(v)
+ }
+ return out
+ }
+ bearer := func(v string) *http.Request {
+ req := httptest.NewRequest("POST", "/x", nil)
+ req.Header.Set("Authorization", "Bearer "+v)
+ return req
+ }
+
+ t.Run("no tokens configured = open (dev mode)", func(t *testing.T) {
+ gate := newTokenGate(nil)
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, httptest.NewRequest("POST", "/x", nil), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("open mode must pass, got %d", rec.Code)
+ }
+ })
+
+ t.Run("valid token passes, wrong/missing rejected", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("s3cret"))
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, bearer("s3cret"), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("valid token must pass, got %d", rec.Code)
+ }
+ rec2 := httptest.NewRecorder()
+ gate(handler)(rec2, bearer("wrong"), nil)
+ if rec2.Code != http.StatusUnauthorized {
+ t.Fatalf("wrong token must 401, got %d", rec2.Code)
+ }
+ rec3 := httptest.NewRecorder()
+ gate(handler)(rec3, httptest.NewRequest("POST", "/x", nil), nil)
+ if rec3.Code != http.StatusUnauthorized {
+ t.Fatalf("missing header must 401, got %d", rec3.Code)
+ }
+ })
+
+ t.Run("multiple tokens accepted (rotation window)", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("old", "new"))
+ for _, tok := range []string{"old", "new"} {
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, bearer(tok), nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("token %q must pass, got %d", tok, rec.Code)
+ }
+ }
+ })
+
+ t.Run("X-API-Token header also accepted", func(t *testing.T) {
+ gate := newTokenGate(mkTokens("s3cret"))
+ req := httptest.NewRequest("POST", "/x", nil)
+ req.Header.Set("X-API-Token", "s3cret")
+ rec := httptest.NewRecorder()
+ gate(handler)(rec, req, nil)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("X-API-Token must pass, got %d", rec.Code)
+ }
+ })
+
+ t.Run("validateStaticToken matches configured list only", func(t *testing.T) {
+ staticTokens = []string{"alpha", "beta"}
+ if !validateStaticToken("beta") {
+ t.Fatal("beta must validate")
+ }
+ if validateStaticToken("gamma") {
+ t.Fatal("gamma must not validate")
+ }
+ if validateStaticToken("") {
+ t.Fatal("empty must not validate")
+ }
+ })
+}
+
+func TestHashTokenOneWay(t *testing.T) {
+ h1 := hashToken("my-token")
+ if h1 == "my-token" || len(h1) != 64 { // sha256 hex
+ t.Fatalf("hash must be 64-hex and differ from input, got %q", h1)
+ }
+ if hashToken("my-token") != h1 {
+ t.Fatal("hash must be deterministic")
+ }
+ if hashToken("my-token2") == h1 {
+ t.Fatal("different inputs must hash differently")
+ }
+}
diff --git a/modules/easysearch/cluster_api.go b/modules/easysearch/cluster_api.go
index e757610a1..7b8845737 100644
--- a/modules/easysearch/cluster_api.go
+++ b/modules/easysearch/cluster_api.go
@@ -4,6 +4,7 @@ package easysearch
import (
"errors"
+ "fmt"
"net/http"
"infini.sh/framework/core/api"
@@ -12,6 +13,7 @@ import (
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/security"
"infini.sh/framework/core/util"
+ "infini.sh/framework/modules/elastic/common"
)
// ──────────────────────────────────────────────────────────────────────────
@@ -97,6 +99,27 @@ func registerClusterAPI() {
}
return nil
},
+ // Live registration: writing a cluster record takes effect
+ // immediately — no restart or boot-time ORM reload needed. This is
+ // what lets a manager (e.g. LogPilot) push sink clusters to gateways
+ // dynamically; pipelines referencing the cluster id resolve on the
+ // next use.
+ PostCreate: func(cfg *elastic.ElasticsearchConfig) error {
+ if _, err := common.InitElasticInstance(*cfg); err != nil {
+ return fmt.Errorf("cluster %s saved but live registration failed: %w", cfg.ID, err)
+ }
+ return nil
+ },
+ PostUpdate: func(cfg *elastic.ElasticsearchConfig) error {
+ if _, err := common.InitElasticInstance(*cfg); err != nil {
+ return fmt.Errorf("cluster %s updated but live re-registration failed: %w", cfg.ID, err)
+ }
+ return nil
+ },
+ PostDelete: func(cfg *elastic.ElasticsearchConfig) error {
+ elastic.RemoveInstance(cfg.ID)
+ return nil
+ },
})
// Pre-registration connectivity probe — deliberately bespoke.
diff --git a/modules/keystore/api/api.go b/modules/keystore/api/api.go
index afda3f87e..fde85be24 100644
--- a/modules/keystore/api/api.go
+++ b/modules/keystore/api/api.go
@@ -33,7 +33,9 @@ import (
httprouter "infini.sh/framework/core/api/router"
"infini.sh/framework/core/keystore"
"infini.sh/framework/core/util"
+ kslib "infini.sh/framework/lib/keystore"
"net/http"
+ "sort"
)
type APIHandler struct {
@@ -77,3 +79,49 @@ func (h *APIHandler) setKeystoreValue(w http.ResponseWriter, req *http.Request,
"success": true,
}, http.StatusOK)
}
+
+
+// listKeystoreKeys — GET /keystore — key names only. Values are secrets:
+// the API is write-only for them by design (nothing can read them back).
+func (h *APIHandler) listKeystoreKeys(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ ks, err := keystore.GetOrInitKeystore()
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ listKs, err := kslib.AsListingKeystore(ks)
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ keys, err := listKs.List()
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ sort.Strings(keys)
+ h.WriteJSON(w, util.MapStr{"keys": keys}, http.StatusOK)
+}
+
+// deleteKeystoreKey — DELETE /keystore?key=...
+func (h *APIHandler) deleteKeystoreKey(w http.ResponseWriter, req *http.Request, _ httprouter.Params) {
+ key := req.URL.Query().Get("key")
+ if key == "" {
+ h.WriteError(w, "key cannot be empty", http.StatusBadRequest)
+ return
+ }
+ ks, err := keystore.GetWriteableKeystore()
+ if err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ if err := ks.Delete(key); err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ if err := ks.Save(); err != nil {
+ h.WriteError(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ h.WriteJSON(w, util.MapStr{"deleted": true, "key": key}, http.StatusOK)
+}
diff --git a/modules/keystore/api/init.go b/modules/keystore/api/init.go
index 00171fb13..af42d4fba 100644
--- a/modules/keystore/api/init.go
+++ b/modules/keystore/api/init.go
@@ -32,4 +32,6 @@ import "infini.sh/framework/core/api"
func Init() {
handler := APIHandler{}
api.HandleAPIMethod(api.POST, "/keystore", handler.setKeystoreValue)
+ api.HandleAPIMethod(api.GET, "/keystore", handler.listKeystoreKeys)
+ api.HandleAPIMethod(api.DELETE, "/keystore", handler.deleteKeystoreKey)
}
diff --git a/modules/pipeline/for_each.go b/modules/pipeline/for_each.go
index 08cc24e73..b972b890f 100644
--- a/modules/pipeline/for_each.go
+++ b/modules/pipeline/for_each.go
@@ -206,6 +206,8 @@ func (p *ForEachProcessor) Process(c *pipeline.Context) error {
continue
}
c.Set(pipeline.RecordContextKey, rec)
+ clones := &[]*event.Event{}
+ c.Set(pipeline.CloneContextKey, clones)
if p.cfg.OnFailure == "tag" {
var tags []string
c.Set(pipeline.FailureTagsKey, &tags)
@@ -249,7 +251,27 @@ func (p *ForEachProcessor) Process(c *pipeline.Context) error {
i := decodedIdx[j]
msgs[i].Data = encoded
msgs[i].Size = len(encoded)
+
+ // Clones (clone processor): materialize each as an extra batch
+ // member sharing the original's offset semantics.
+ for _, cl := range pipeline.TakeClones(c) {
+ if pipeline.IsDropped(cl) {
+ continue
+ }
+ cloned, cerr := p.codec.Encode(cl)
+ if cerr != nil {
+ log.Warnf("for_each: failed to encode clone at offset %v: %v", msgs[i].Offset, cerr)
+ continue
+ }
+ msgs = append(msgs, queue.Message{
+ Offset: msgs[i].Offset,
+ Data: cloned,
+ Size: len(cloned),
+ })
+ }
}
+ // publish the (possibly extended) batch back to the context
+ c.Set(param.ParaKey(p.cfg.MessageField), msgs)
return nil
}
diff --git a/modules/security/access_token/authentication.go b/modules/security/access_token/authentication.go
index a1c860ee8..4a9057553 100644
--- a/modules/security/access_token/authentication.go
+++ b/modules/security/access_token/authentication.go
@@ -16,6 +16,7 @@ import (
"infini.sh/framework/core/api"
httprouter "infini.sh/framework/core/api/router"
+ "infini.sh/framework/core/elastic"
"infini.sh/framework/core/errors"
"infini.sh/framework/core/global"
"infini.sh/framework/core/kv"
@@ -609,3 +610,79 @@ func listAccessTokensFromKV(ownerID string) ([]util.MapStr, error) {
}
return out, nil
}
+
+// ListTokens returns all access tokens (KV mode reads the id index; native
+// mode reads the ORM). Sensitive: the plaintext token strings are included —
+// admin-surface only.
+func ListTokens() ([]*security.AccessToken, error) {
+ if isNative() {
+ ctx := orm.NewContext().DirectAccess()
+ ctx.PermissionScope(security.PermissionScopePlatform)
+ res, err := orm.SearchV2(ctx, orm.NewQuery().Size(1000))
+ if err != nil {
+ return nil, err
+ }
+ rows, _, err := elastic.DecodeHits[security.AccessToken](res)
+ out := make([]*security.AccessToken, 0, len(rows))
+ for i := range rows {
+ out = append(out, &rows[i])
+ }
+ return out, err
+ }
+ ids, err := loadTokenIDs()
+ if err != nil {
+ return nil, err
+ }
+ out := make([]*security.AccessToken, 0, len(ids))
+ for _, id := range ids {
+ tokenString, err := kv.GetValue(kvAccessTokenIndexBucket, []byte(id))
+ if err != nil || len(tokenString) == 0 {
+ continue
+ }
+ t, err := GetToken(string(tokenString))
+ if err != nil || t == nil {
+ continue
+ }
+ out = append(out, t)
+ }
+ return out, nil
+}
+
+// DeleteTokenByID revokes a token by its id (removes the KV record and the
+// id index entry; native mode deletes the ORM row). The presented token
+// stops validating immediately.
+func DeleteTokenByID(tokenID string) error {
+ if tokenID == "" {
+ return fmt.Errorf("token id is required")
+ }
+ if isNative() {
+ ctx := orm.NewContext().DirectAccess()
+ ctx.PermissionScope(security.PermissionScopePlatform)
+ t := security.AccessToken{}
+ t.ID = tokenID
+ return orm.Delete(ctx, &t)
+ }
+ tokenString, err := kv.GetValue(kvAccessTokenIndexBucket, []byte(tokenID))
+ if err != nil {
+ return err
+ }
+ if len(tokenString) > 0 {
+ if err := kv.DeleteKey(KVAccessTokenBucket, tokenString); err != nil {
+ return err
+ }
+ }
+ if err := kv.DeleteKey(kvAccessTokenIndexBucket, []byte(tokenID)); err != nil {
+ return err
+ }
+ ids, err := loadTokenIDs()
+ if err != nil {
+ return err
+ }
+ rest := make([]string, 0, len(ids))
+ for _, id := range ids {
+ if id != tokenID {
+ rest = append(rest, id)
+ }
+ }
+ return saveTokenIDs(rest)
+}
diff --git a/plugins/elastic/indexing_merge/indexing_merge.go b/plugins/elastic/indexing_merge/indexing_merge.go
index 57b6ec16f..a6785bc4b 100644
--- a/plugins/elastic/indexing_merge/indexing_merge.go
+++ b/plugins/elastic/indexing_merge/indexing_merge.go
@@ -287,6 +287,15 @@ READ_DOCS:
util.WalkBytesAndReplace(pop, util.NEWLINE, util.SPACE)
+ // Data stream (create) documents must carry a top-level date
+ // @timestamp; the otel envelope keeps the timestamp at the
+ // envelope level and the record body under "payload", which
+ // neither reaches the doc top level. Normalize envelope docs
+ // for data stream semantics; bare docs pass through as-is.
+ if writeOpType == "create" {
+ pop = normalizeDataStreamDoc(pop)
+ }
+
docBuf.Write(pop)
docBuf.WriteString("\n")
@@ -342,3 +351,70 @@ CLEAN_BUFFER:
}
goto READ_DOCS
}
+
+// normalizeDataStreamDoc turns one otel envelope (the queue's LogEvent JSON:
+// {"metadata":{...},"payload":{...},"timestamp":"..."}) into a data stream
+// compatible document:
+//
+// - payload fields are promoted to the top level (message etc. become
+// first-class doc fields)
+// - a top-level @timestamp is derived from payload.timestamp /
+// payload.observed_timestamp / the envelope timestamp / now — data
+// streams reject documents without a date @timestamp
+// - timestamp mirrors @timestamp (conventional sort field)
+// - metadata.file (agent collection origin) is kept as top-level "file"
+//
+// Documents that are not envelopes (no "payload" key) pass through
+// unchanged except for the @timestamp stamp.
+func normalizeDataStreamDoc(doc []byte) []byte {
+ var m util.MapStr
+ if err := util.FromJSONBytes(doc, &m); err != nil {
+ return doc
+ }
+
+ ts := ""
+ if payload, ok := m["payload"].(map[string]interface{}); ok {
+ for _, k := range []string{"timestamp", "@timestamp", "observed_timestamp"} {
+ if v, ok := payload[k].(string); ok && v != "" {
+ ts = v
+ break
+ }
+ }
+ }
+ if ts == "" {
+ if v, ok := m["timestamp"].(string); ok {
+ ts = v
+ }
+ }
+ if ts == "" {
+ ts = time.Now().UTC().Format(time.RFC3339Nano)
+ }
+
+ out := util.MapStr{}
+ if payload, ok := m["payload"].(map[string]interface{}); ok {
+ for k, v := range payload {
+ out[k] = v
+ }
+ } else {
+ for k, v := range m {
+ out[k] = v
+ }
+ }
+ out["@timestamp"] = ts
+ out["timestamp"] = ts
+
+ if meta, ok := m["metadata"].(map[string]interface{}); ok {
+ if f, ok := meta["file"].(map[string]interface{}); ok {
+ out["file"] = f
+ }
+ if rt, ok := meta["log_type"].(string); ok && rt != "" {
+ out["log_type"] = rt
+ }
+ // per-pattern label (e.g. server|deprecation|slowlog|gc) — virtual
+ // streams split on this field.
+ if lk, ok := meta["log_kind"].(string); ok && lk != "" {
+ out["log_kind"] = lk
+ }
+ }
+ return util.MustToJSONBytes(out)
+}
diff --git a/plugins/elastic/indexing_merge/indexing_merge_test.go b/plugins/elastic/indexing_merge/indexing_merge_test.go
new file mode 100644
index 000000000..00e5d2a87
--- /dev/null
+++ b/plugins/elastic/indexing_merge/indexing_merge_test.go
@@ -0,0 +1,82 @@
+/* ©INFINI, All Rights Reserved. */
+
+package indexing_merge
+
+import (
+ "encoding/json"
+ "testing"
+
+ "infini.sh/framework/core/util"
+)
+
+// TestNormalizeDataStreamDoc verifies the otel-envelope → data stream doc
+// normalization: payload promotion, @timestamp derivation priority, and
+// metadata promotion (file / log_type / log_kind).
+func TestNormalizeDataStreamDoc(t *testing.T) {
+ envelope := `{"metadata":{"file":{"path":"/var/log/system.log","offset":123},"log_type":"text","log_kind":"server"},
+ "payload":{"message":"adding data stream [system-logs]","observed_timestamp":"2026-08-23T14:38:32.982744Z"},
+ "timestamp":"2026-08-23T14:38:32Z"}`
+
+ out := normalizeDataStreamDoc([]byte(envelope))
+ var doc util.MapStr
+ if err := json.Unmarshal(out, &doc); err != nil {
+ t.Fatalf("unmarshal normalized doc: %v", err)
+ }
+
+ // payload promoted to top level; envelope keys dropped.
+ if doc["message"] != "adding data stream [system-logs]" {
+ t.Fatalf("message not promoted: %v", doc)
+ }
+ if _, ok := doc["payload"]; ok {
+ t.Fatalf("payload key should not survive: %v", doc)
+ }
+
+ // @timestamp: payload 无显式 timestamp 时 observed_timestamp 优先于信封级;
+ // timestamp 镜像之。
+ if doc["@timestamp"] != "2026-08-23T14:38:32.982744Z" || doc["timestamp"] != "2026-08-23T14:38:32.982744Z" {
+ t.Fatalf("timestamp fields: @=%v ts=%v", doc["@timestamp"], doc["timestamp"])
+ }
+
+ // metadata promotion.
+ file, ok := doc["file"].(map[string]interface{})
+ if !ok || file["path"] != "/var/log/system.log" {
+ t.Fatalf("file metadata not promoted: %v", doc["file"])
+ }
+ if doc["log_type"] != "text" || doc["log_kind"] != "server" {
+ t.Fatalf("label fields not promoted: %v", doc)
+ }
+}
+
+// TestNormalizeDataStreamDocTimestampPriority: payload.timestamp wins over
+// observed_timestamp and the envelope timestamp.
+func TestNormalizeDataStreamDocTimestampPriority(t *testing.T) {
+ for _, tc := range []struct {
+ name, payload, envelope, want string
+ }{
+ {"payload timestamp first", `"timestamp":"2026-01-01T00:00:00Z","observed_timestamp":"2026-02-02T00:00:00Z"`, "2026-03-03T00:00:00Z", "2026-01-01T00:00:00Z"},
+ {"observed over envelope", `"observed_timestamp":"2026-02-02T00:00:00Z"`, "2026-03-03T00:00:00Z", "2026-02-02T00:00:00Z"},
+ {"envelope fallback", `"unused":"x"`, "2026-03-03T00:00:00Z", "2026-03-03T00:00:00Z"},
+ } {
+ env := `{"payload":{` + tc.payload + `},"timestamp":"` + tc.envelope + `"}`
+ out := normalizeDataStreamDoc([]byte(env))
+ var doc util.MapStr
+ _ = json.Unmarshal(out, &doc)
+ if doc["@timestamp"] != tc.want {
+ t.Fatalf("%s: @timestamp = %v, want %v", tc.name, doc["@timestamp"], tc.want)
+ }
+ }
+}
+
+// TestNormalizeDataStreamDocBareDoc: non-envelope docs keep their fields
+// and only get the @timestamp stamp.
+func TestNormalizeDataStreamDocBareDoc(t *testing.T) {
+ out := normalizeDataStreamDoc([]byte(`{"message":"raw","level":"info"}`))
+ var doc util.MapStr
+ _ = json.Unmarshal(out, &doc)
+ if doc["message"] != "raw" || doc["level"] != "info" {
+ t.Fatalf("bare doc fields lost: %v", doc)
+ }
+ if doc["timestamp"] == "" || doc["@timestamp"] == "" {
+ t.Fatalf("bare doc missing timestamp stamps: %v", doc)
+ }
+}
diff --git a/plugins/queue/queue_output/queue_output.go b/plugins/queue/queue_output/queue_output.go
new file mode 100644
index 000000000..128522c58
--- /dev/null
+++ b/plugins/queue/queue_output/queue_output.go
@@ -0,0 +1,81 @@
+/* Copyright © INFINI Ltd. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+// Package queue_output provides the "queue_output" pipeline processor: the
+// chain-tail companion of "consumer". It takes the message batch the
+// consumer exposed in the context (typically after a for_each transform
+// chain) and appends every record onto a target queue, enabling two-stage
+// pipelines: process on one queue, sink (e.g. bulk_indexing) from another.
+//
+// Configuration:
+//
+// - queue_output:
+// queue_name: indexing-my-stream # target queue (required)
+// message_field: messages # ctx batch key (default "messages")
+//
+// On queue push failure the processor returns an error so the consumer
+// does not commit the offset and the batch is redelivered (at-least-once).
+package queue_output
+
+import (
+ "fmt"
+
+ "infini.sh/framework/core/config"
+ "infini.sh/framework/core/param"
+ "infini.sh/framework/core/pipeline"
+ "infini.sh/framework/core/queue"
+ log "infini.sh/framework/core/log"
+)
+
+const name = "queue_output"
+
+type Config struct {
+ QueueName string `config:"queue_name"`
+ MessageField string `config:"message_field"`
+}
+
+type Processor struct {
+ cfg Config
+}
+
+func init() {
+ pipeline.RegisterProcessorPlugin(name, New)
+}
+
+func New(c *config.Config) (pipeline.Processor, error) {
+ cfg := Config{MessageField: "messages"}
+ if err := c.Unpack(&cfg); err != nil {
+ return nil, fmt.Errorf("failed to unpack the configuration of %s processor: %s", name, err)
+ }
+ if cfg.QueueName == "" {
+ return nil, fmt.Errorf("%s processor requires queue_name", name)
+ }
+ return &Processor{cfg: cfg}, nil
+}
+
+func (p *Processor) Name() string { return name }
+
+// Process appends every record of the context batch onto the target queue.
+func (p *Processor) Process(c *pipeline.Context) error {
+ v := c.Get(param.ParaKey(p.cfg.MessageField))
+ msgs, ok := v.([]queue.Message)
+ if !ok || len(msgs) == 0 {
+ return nil
+ }
+
+ qConfig := queue.GetOrInitConfig(p.cfg.QueueName)
+ pushed := 0
+ for i := range msgs {
+ if len(msgs[i].Data) == 0 {
+ continue // dropped records (drop_event) are skipped
+ }
+ if err := queue.Push(qConfig, msgs[i].Data); err != nil {
+ log.Errorf("%s: queue push failed after %d/%d records: %v", name, pushed, len(msgs), err)
+ return fmt.Errorf("%s: queue push failed after %d/%d records: %w",
+ name, pushed, len(msgs), err)
+ }
+ pushed++
+ }
+ return nil
+}