415 lines
11 KiB
Go
415 lines
11 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/x-grk/longpolling-remote-access/internal/proto"
|
|
)
|
|
|
|
// Sending a file or folder to another machine. The sending agent packs it as
|
|
// a tar stream; to another agent the stream goes in pieces through a relay
|
|
// here, a few pieces held at most, and the receiving agent unpacks it as it
|
|
// comes. To an SSH host the sending agent writes it there itself. The panel
|
|
// follows a job: how far the sender is, and the result.
|
|
|
|
// relay carries one stream between two agents, in numbered pieces.
|
|
type relay struct {
|
|
id, from, to string
|
|
|
|
mu sync.Mutex
|
|
changed chan struct{}
|
|
chunks map[int][]byte
|
|
low int // pieces below it were taken
|
|
end int // the last piece's number, -1 until it came
|
|
err string
|
|
}
|
|
|
|
func (x *relay) notify() {
|
|
close(x.changed)
|
|
x.changed = make(chan struct{})
|
|
}
|
|
|
|
// await waits until ready() holds and returns true with x.mu held.
|
|
func (x *relay) await(ctx context.Context, d time.Duration, ready func() bool) bool {
|
|
t := time.NewTimer(d)
|
|
defer t.Stop()
|
|
for {
|
|
x.mu.Lock()
|
|
if ready() {
|
|
return true
|
|
}
|
|
ch := x.changed
|
|
x.mu.Unlock()
|
|
select {
|
|
case <-ch:
|
|
case <-t.C:
|
|
return false
|
|
case <-ctx.Done():
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
func (x *relay) fail(msg string) {
|
|
x.mu.Lock()
|
|
if x.err == "" {
|
|
x.err = msg
|
|
x.chunks = nil
|
|
x.notify()
|
|
}
|
|
x.mu.Unlock()
|
|
}
|
|
|
|
type shipJob struct {
|
|
id string
|
|
mu sync.Mutex
|
|
changed chan struct{}
|
|
ver int
|
|
prog proto.ShipProgress
|
|
done bool
|
|
res proto.TaskResult
|
|
cancel context.CancelFunc
|
|
stopped bool
|
|
}
|
|
|
|
func (j *shipJob) update(fn func()) {
|
|
j.mu.Lock()
|
|
fn()
|
|
j.ver++
|
|
close(j.changed)
|
|
j.changed = make(chan struct{})
|
|
j.mu.Unlock()
|
|
}
|
|
|
|
type shipping struct {
|
|
mu sync.Mutex
|
|
relays map[string]*relay
|
|
jobs map[string]*shipJob
|
|
}
|
|
|
|
func (c *shipping) relay(id string) *relay {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.relays[id]
|
|
}
|
|
|
|
func (c *shipping) job(id string) *shipJob {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.jobs[id]
|
|
}
|
|
|
|
// shipReq is what the panel asks for: Src on this agent to Dir/Name on agent
|
|
// To, or on the SSH host.
|
|
type shipReq struct {
|
|
Src string `json:"src"`
|
|
Name string `json:"name"`
|
|
Dir string `json:"dir"`
|
|
Conflict string `json:"conflict"`
|
|
Move bool `json:"move"`
|
|
To string `json:"to"`
|
|
SSH *proto.SSHTarget `json:"ssh"`
|
|
}
|
|
|
|
// POST /api/panel/agents/{id}/fs/ship - start sending; the answer names the
|
|
// job to follow.
|
|
func (s *Server) handleFSShip(w http.ResponseWriter, r *http.Request) {
|
|
from := r.PathValue("id")
|
|
var q shipReq
|
|
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&q); err != nil {
|
|
writeJSON(w, proto.TaskResult{Error: "bad request"})
|
|
return
|
|
}
|
|
q.Name, q.Dir = strings.TrimSpace(q.Name), strings.TrimSpace(q.Dir)
|
|
if q.Src == "" || q.Name == "" || strings.Contains(q.Name, "/") || q.Dir == "" {
|
|
writeJSON(w, proto.TaskResult{Error: "не указано, что и куда"})
|
|
return
|
|
}
|
|
if !s.fsAgent(w, from, true) {
|
|
return
|
|
}
|
|
where := ""
|
|
if q.SSH != nil {
|
|
t := q.SSH
|
|
t.Host, t.User = strings.TrimSpace(t.Host), strings.TrimSpace(t.User)
|
|
if t.Port == 0 {
|
|
t.Port = 22
|
|
}
|
|
if t.Host == "" || t.User == "" || t.Port < 1 || t.Port > 65535 {
|
|
writeJSON(w, proto.TaskResult{Error: "нужны хост, порт и пользователь"})
|
|
return
|
|
}
|
|
where = "ssh:" + t.User + "@" + t.Host + ":" + strconv.Itoa(t.Port)
|
|
} else {
|
|
if q.To == "" {
|
|
writeJSON(w, proto.TaskResult{Error: "не выбрана машина"})
|
|
return
|
|
}
|
|
if !s.fsAgent(w, q.To, true) {
|
|
return
|
|
}
|
|
where = "agent:" + q.To
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), xferTimeout)
|
|
j := &shipJob{id: randToken()[:24], changed: make(chan struct{}), cancel: cancel}
|
|
s.ship.mu.Lock()
|
|
if s.ship.jobs == nil {
|
|
s.ship.jobs, s.ship.relays = map[string]*shipJob{}, map[string]*relay{}
|
|
}
|
|
s.ship.jobs[j.id] = j
|
|
s.ship.mu.Unlock()
|
|
if admin := s.currentAdmin(r); admin != nil {
|
|
op := "copy"
|
|
if q.Move {
|
|
op = "move"
|
|
}
|
|
s.db.Audit(admin.ID, from, "fs.ship."+op+":"+q.Src+" -> "+where+":"+strings.TrimSuffix(q.Dir, "/")+"/"+q.Name)
|
|
}
|
|
go s.runShip(ctx, j, from, q)
|
|
data, _ := json.Marshal(map[string]string{"job": j.id})
|
|
writeJSON(w, proto.TaskResult{OK: true, Data: data})
|
|
}
|
|
|
|
func (s *Server) runShip(ctx context.Context, j *shipJob, from string, q shipReq) {
|
|
defer j.cancel()
|
|
res := s.ship1(ctx, j, from, q)
|
|
if !res.OK && j.stopped {
|
|
res.Error = "отменено"
|
|
}
|
|
j.update(func() { j.done, j.res = true, res })
|
|
// The result waits a while for the panel to come for it.
|
|
time.AfterFunc(xferIdle, func() {
|
|
s.ship.mu.Lock()
|
|
delete(s.ship.jobs, j.id)
|
|
s.ship.mu.Unlock()
|
|
})
|
|
}
|
|
|
|
func oldAgent(res proto.TaskResult) proto.TaskResult {
|
|
if unknownOp(res) {
|
|
res.Error = "агент не умеет отправлять на другие машины - обновите его"
|
|
}
|
|
return res
|
|
}
|
|
|
|
func (s *Server) ship1(ctx context.Context, j *shipJob, from string, q shipReq) proto.TaskResult {
|
|
args := proto.FSShipArgs{Src: q.Src, Name: q.Name, Dir: q.Dir, Conflict: q.Conflict, Job: j.id}
|
|
if q.SSH != nil {
|
|
args.SSH, args.Move = q.SSH, q.Move
|
|
res, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout)
|
|
if err != nil {
|
|
return proto.TaskResult{Error: err.Error()}
|
|
}
|
|
return oldAgent(res)
|
|
}
|
|
|
|
x := &relay{id: randToken()[:24], from: from, to: q.To, changed: make(chan struct{}), chunks: map[int][]byte{}, end: -1}
|
|
s.ship.mu.Lock()
|
|
s.ship.relays[x.id] = x
|
|
s.ship.mu.Unlock()
|
|
defer func() {
|
|
s.ship.mu.Lock()
|
|
delete(s.ship.relays, x.id)
|
|
s.ship.mu.Unlock()
|
|
}()
|
|
go func() { <-ctx.Done(); x.fail("отменено") }()
|
|
args.Relay, args.Chunk = x.id, xferChunk
|
|
|
|
var in proto.TaskResult
|
|
inDone := make(chan struct{})
|
|
go func() {
|
|
defer close(inDone)
|
|
res, err := s.waitTask(ctx, q.To, proto.TaskFSShipIn, args, xferTimeout)
|
|
if err != nil {
|
|
res = proto.TaskResult{Error: err.Error()}
|
|
}
|
|
in = oldAgent(res)
|
|
if !in.OK {
|
|
x.fail(in.Error)
|
|
}
|
|
}()
|
|
out, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout)
|
|
if err != nil {
|
|
out = proto.TaskResult{Error: err.Error()}
|
|
}
|
|
out = oldAgent(out)
|
|
if !out.OK {
|
|
x.fail(out.Error)
|
|
}
|
|
<-inDone
|
|
if !in.OK {
|
|
return proto.TaskResult{Error: "на той машине: " + in.Error}
|
|
}
|
|
if !out.OK {
|
|
return out
|
|
}
|
|
if q.Move {
|
|
res, err := s.waitTask(ctx, from, proto.TaskFSDelTree, proto.FSPathArgs{Path: q.Src}, fsTaskTimeout)
|
|
if err == nil && !res.OK {
|
|
err = errText(res.Error)
|
|
}
|
|
if err != nil {
|
|
return proto.TaskResult{Error: "отправлено, но здесь не удалилось: " + err.Error()}
|
|
}
|
|
}
|
|
// The receiver's count: what it put down and what it kept as it was.
|
|
return in
|
|
}
|
|
|
|
type errText string
|
|
|
|
func (e errText) Error() string { return string(e) }
|
|
|
|
// GET /api/panel/fs/job/{jid}?v=N - the job's state once it differs from
|
|
// version N, or after a while anyway.
|
|
func (s *Server) handleShipJob(w http.ResponseWriter, r *http.Request) {
|
|
j := s.ship.job(r.PathValue("jid"))
|
|
if j == nil {
|
|
writeJSON(w, map[string]any{"done": true, "result": proto.TaskResult{Error: "задание не найдено"}})
|
|
return
|
|
}
|
|
seen, _ := strconv.Atoi(r.URL.Query().Get("v"))
|
|
t := time.NewTimer(25 * time.Second)
|
|
defer t.Stop()
|
|
for {
|
|
j.mu.Lock()
|
|
if j.ver != seen || j.done {
|
|
out := map[string]any{"v": j.ver, "progress": j.prog, "done": j.done}
|
|
if j.done {
|
|
out["result"] = j.res
|
|
}
|
|
j.mu.Unlock()
|
|
writeJSON(w, out)
|
|
return
|
|
}
|
|
ch := j.changed
|
|
j.mu.Unlock()
|
|
select {
|
|
case <-ch:
|
|
case <-t.C:
|
|
seen = -1 // answer with what there is
|
|
case <-r.Context().Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// POST /api/panel/fs/job/{jid}/cancel
|
|
func (s *Server) handleShipCancel(w http.ResponseWriter, r *http.Request) {
|
|
if j := s.ship.job(r.PathValue("jid")); j != nil {
|
|
j.update(func() { j.stopped = true })
|
|
j.cancel()
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// POST /api/agent/fs/job/{jid}/progress - the sender's progress. 410: the
|
|
// job is over or cancelled, stop.
|
|
func (s *Server) handleShipProgress(w http.ResponseWriter, r *http.Request) {
|
|
if s.authenticateAgent(r) == nil {
|
|
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
j := s.ship.job(r.PathValue("jid"))
|
|
var p proto.ShipProgress
|
|
if j == nil || json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&p) != nil {
|
|
http.Error(w, "gone", http.StatusGone)
|
|
return
|
|
}
|
|
stop := false
|
|
j.update(func() { j.prog, stop = p, j.stopped || j.done })
|
|
if stop {
|
|
http.Error(w, "gone", http.StatusGone)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// POST /api/agent/fs/relay/{rid}/{i} - piece i of the stream from the
|
|
// sender, X-Last on the last. 503: no room yet, again; 410: it is over.
|
|
func (s *Server) handleRelayPost(w http.ResponseWriter, r *http.Request) {
|
|
a := s.authenticateAgent(r)
|
|
if a == nil {
|
|
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
x := s.ship.relay(r.PathValue("rid"))
|
|
i, err := strconv.Atoi(r.PathValue("i"))
|
|
if x == nil || x.from != a.ID || err != nil || i < 0 {
|
|
http.Error(w, "передача не найдена", http.StatusGone)
|
|
return
|
|
}
|
|
data, err := readPiece(w, r)
|
|
if err != nil {
|
|
http.Error(w, "кусок не дошёл", http.StatusBadRequest)
|
|
return
|
|
}
|
|
if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || i < x.low+xferBuffer }) {
|
|
http.Error(w, "нет места, повторите", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
defer x.mu.Unlock()
|
|
if x.err != "" {
|
|
http.Error(w, x.err, http.StatusGone)
|
|
return
|
|
}
|
|
if i >= x.low && x.chunks[i] == nil {
|
|
x.chunks[i] = data
|
|
if r.Header.Get("X-Last") != "" {
|
|
x.end = i
|
|
}
|
|
x.notify()
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// GET /api/agent/fs/relay/{rid}/{i} - piece i for the receiver, X-Last on
|
|
// the last. Asking for i says every piece below it is written, so those are
|
|
// let go. 503: not here yet, ask again; 410: it is over.
|
|
func (s *Server) handleRelayGet(w http.ResponseWriter, r *http.Request) {
|
|
a := s.authenticateAgent(r)
|
|
if a == nil {
|
|
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
x := s.ship.relay(r.PathValue("rid"))
|
|
i, err := strconv.Atoi(r.PathValue("i"))
|
|
if x == nil || x.to != a.ID || err != nil || i < 0 {
|
|
http.Error(w, "передача не найдена", http.StatusGone)
|
|
return
|
|
}
|
|
x.mu.Lock()
|
|
if i > x.low {
|
|
for k := x.low; k < i; k++ {
|
|
delete(x.chunks, k)
|
|
}
|
|
x.low = i
|
|
x.notify()
|
|
}
|
|
x.mu.Unlock()
|
|
if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || x.chunks[i] != nil }) {
|
|
http.Error(w, "кусок ещё не пришёл", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
if x.err != "" {
|
|
msg := x.err
|
|
x.mu.Unlock()
|
|
http.Error(w, msg, http.StatusGone)
|
|
return
|
|
}
|
|
data, last := x.chunks[i], i == x.end
|
|
x.mu.Unlock()
|
|
w.Header().Set("Content-Type", "application/octet-stream")
|
|
w.Header().Set("Content-Length", strconv.Itoa(len(data)))
|
|
if last {
|
|
w.Header().Set("X-Last", "1")
|
|
}
|
|
_, _ = w.Write(data)
|
|
}
|