mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-10-07 02:06:38 +08:00
feat(federation): add remote node command support
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
@@ -77,6 +77,18 @@ type RuntimeDiagnoseRequest struct {
|
|||||||
Timeout int `json:"timeout"`
|
Timeout int `json:"timeout"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type RuntimeNodeCommandRequest struct {
|
||||||
|
CommandType string `json:"commandType"`
|
||||||
|
Data interface{} `json:"data"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type RuntimeNodeCommandResponse struct {
|
||||||
|
Type string `json:"type"`
|
||||||
|
Success bool `json:"success"`
|
||||||
|
Message string `json:"message"`
|
||||||
|
Data map[string]interface{} `json:"data,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
func NewFederationClient() *FederationClient {
|
func NewFederationClient() *FederationClient {
|
||||||
return &FederationClient{
|
return &FederationClient{
|
||||||
client: &http.Client{
|
client: &http.Client{
|
||||||
@@ -333,3 +345,42 @@ func (c *FederationClient) Diagnose(url, token, localDomain string, reqData Runt
|
|||||||
|
|
||||||
return res.Data, nil
|
return res.Data, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (c *FederationClient) Command(url, token, localDomain string, reqData RuntimeNodeCommandRequest) (*RuntimeNodeCommandResponse, error) {
|
||||||
|
url = strings.TrimSuffix(url, "/")
|
||||||
|
bodyBytes, _ := json.Marshal(reqData)
|
||||||
|
req, err := http.NewRequest("POST", url+"/api/v1/federation/runtime/command", strings.NewReader(string(bodyBytes)))
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
req.Header.Set("Authorization", "Bearer "+token)
|
||||||
|
if localDomain != "" {
|
||||||
|
req.Header.Set("X-Panel-Domain", localDomain)
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
|
||||||
|
resp, err := c.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
if resp.StatusCode != 200 {
|
||||||
|
body, _ := io.ReadAll(resp.Body)
|
||||||
|
return nil, fmt.Errorf("remote error %d: %s", resp.StatusCode, string(body))
|
||||||
|
}
|
||||||
|
|
||||||
|
var res struct {
|
||||||
|
Code int `json:"code"`
|
||||||
|
Msg string `json:"msg"`
|
||||||
|
Data RuntimeNodeCommandResponse `json:"data"`
|
||||||
|
}
|
||||||
|
if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
if res.Code != 0 {
|
||||||
|
return nil, fmt.Errorf("remote api error: %s", res.Msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &res.Data, nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -91,6 +91,11 @@ type federationRuntimeDiagnoseRequest struct {
|
|||||||
Timeout int `json:"timeout"`
|
Timeout int `json:"timeout"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type federationRuntimeCommandRequest struct {
|
||||||
|
CommandType string `json:"commandType"`
|
||||||
|
Data interface{} `json:"data"`
|
||||||
|
}
|
||||||
|
|
||||||
type peerShareUsedPort struct {
|
type peerShareUsedPort struct {
|
||||||
RuntimeID int64 `json:"runtimeId"`
|
RuntimeID int64 `json:"runtimeId"`
|
||||||
Port int `json:"port"`
|
Port int `json:"port"`
|
||||||
@@ -1199,6 +1204,51 @@ func (h *Handler) federationRuntimeDiagnose(w http.ResponseWriter, r *http.Reque
|
|||||||
response.WriteJSON(w, response.OK(res.Data))
|
response.WriteJSON(w, response.OK(res.Data))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (h *Handler) federationRuntimeCommand(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if r.Method != http.MethodPost {
|
||||||
|
response.WriteJSON(w, response.ErrDefault("Invalid method"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
token := extractBearerToken(r)
|
||||||
|
share, err := h.repo.GetPeerShareByToken(token)
|
||||||
|
if err != nil || share == nil {
|
||||||
|
response.WriteJSON(w, response.Err(401, "Unauthorized"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var req federationRuntimeCommandRequest
|
||||||
|
if err := decodeJSON(r.Body, &req); err != nil {
|
||||||
|
response.WriteJSON(w, response.ErrDefault("Invalid JSON"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
cmd := strings.TrimSpace(req.CommandType)
|
||||||
|
if cmd == "" {
|
||||||
|
response.WriteJSON(w, response.ErrDefault("commandType is required"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !isFederationRuntimeCommandAllowed(cmd) {
|
||||||
|
response.WriteJSON(w, response.ErrDefault("command not allowed"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err := h.sendNodeCommand(share.NodeID, cmd, req.Data, false, false)
|
||||||
|
if err != nil {
|
||||||
|
response.WriteJSON(w, response.ErrDefault(err.Error()))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
response.WriteJSON(w, response.OK(res))
|
||||||
|
}
|
||||||
|
|
||||||
|
func isFederationRuntimeCommandAllowed(commandType string) bool {
|
||||||
|
switch strings.ToLower(strings.TrimSpace(commandType)) {
|
||||||
|
case "addservice", "updateservice", "deleteservice", "pauseservice", "resumeservice", "addchains", "deletechains", "addlimiters", "deletelimiters", "tcpping", "reload":
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (h *Handler) pickPeerSharePort(share *sqlite.PeerShare, requestedPort int) (int, error) {
|
func (h *Handler) pickPeerSharePort(share *sqlite.PeerShare, requestedPort int) (int, error) {
|
||||||
if share == nil {
|
if share == nil {
|
||||||
return 0, fmt.Errorf("share not found")
|
return 0, fmt.Errorf("share not found")
|
||||||
|
|||||||
@@ -169,6 +169,7 @@ func (h *Handler) Register(mux *http.ServeMux) {
|
|||||||
mux.HandleFunc("/api/v1/federation/runtime/apply-role", h.authPeer(h.federationRuntimeApplyRole))
|
mux.HandleFunc("/api/v1/federation/runtime/apply-role", h.authPeer(h.federationRuntimeApplyRole))
|
||||||
mux.HandleFunc("/api/v1/federation/runtime/release-role", h.authPeer(h.federationRuntimeReleaseRole))
|
mux.HandleFunc("/api/v1/federation/runtime/release-role", h.authPeer(h.federationRuntimeReleaseRole))
|
||||||
mux.HandleFunc("/api/v1/federation/runtime/diagnose", h.authPeer(h.federationRuntimeDiagnose))
|
mux.HandleFunc("/api/v1/federation/runtime/diagnose", h.authPeer(h.federationRuntimeDiagnose))
|
||||||
|
mux.HandleFunc("/api/v1/federation/runtime/command", h.authPeer(h.federationRuntimeCommand))
|
||||||
mux.HandleFunc("/api/v1/federation/node/import", h.nodeImport)
|
mux.HandleFunc("/api/v1/federation/node/import", h.nodeImport)
|
||||||
|
|
||||||
mux.HandleFunc("/flow/test", h.flowTest)
|
mux.HandleFunc("/flow/test", h.flowTest)
|
||||||
|
|||||||
@@ -93,6 +93,8 @@ func shouldSkip(path string) bool {
|
|||||||
return true
|
return true
|
||||||
case path == "/api/v1/federation/runtime/diagnose":
|
case path == "/api/v1/federation/runtime/diagnose":
|
||||||
return true
|
return true
|
||||||
|
case path == "/api/v1/federation/runtime/command":
|
||||||
|
return true
|
||||||
default:
|
default:
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user