diff --git a/VERSION.txt b/VERSION.txt index 81340c7..bbdeab6 100644 --- a/VERSION.txt +++ b/VERSION.txt @@ -1 +1 @@ -0.0.4 +0.0.5 diff --git a/go.mod b/go.mod index 711c290..97d7dd6 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26 require ( github.com/creack/pty v1.1.21 golang.org/x/crypto v0.31.0 + golang.org/x/sys v0.28.0 modernc.org/sqlite v1.24.0 ) @@ -13,7 +14,6 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/mattn/go-isatty v0.0.16 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect - golang.org/x/sys v0.28.0 // indirect modernc.org/libc v1.22.5 // indirect modernc.org/mathutil v1.5.0 // indirect modernc.org/memory v1.5.0 // indirect diff --git a/internal/agent/speed.go b/internal/agent/speed.go new file mode 100644 index 0000000..93e6114 --- /dev/null +++ b/internal/agent/speed.go @@ -0,0 +1,122 @@ +package agent + +import ( + "context" + "crypto/rand" + "encoding/json" + "io" + "net/http" + "time" + + "github.com/x-grk/longpolling-remote-access/internal/proto" +) + +const ( + // speedWindow is how long each direction is measured for. + speedWindow = 3 * time.Second + // speedUpCap bounds the upload, whatever the link does in speedWindow. + speedUpCap = 48 << 20 +) + +// doSpeed measures the agent's HTTP link to the server: round-trip time of an +// empty request, then how much flows each way in speedWindow. +func (a *Agent) doSpeed(ctx context.Context) proto.TaskResult { + res, err := a.measureSpeed(ctx) + if err != nil { + return proto.TaskResult{Error: err.Error()} + } + data, _ := json.Marshal(res) + return proto.TaskResult{OK: true, Data: data} +} + +func (a *Agent) measureSpeed(ctx context.Context) (proto.SysSpeedResult, error) { + var res proto.SysSpeedResult + // Ping: the first request may pay for a new connection, so keep the best. + for i := 0; i < 4; i++ { + start := time.Now() + resp, err := a.speedRequest(ctx, http.MethodGet, "/api/agent/speed/down?bytes=0", nil) + if err != nil { + return res, err + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + ms := float64(time.Since(start).Microseconds()) / 1000 + if i == 0 || ms < res.PingMs { + res.PingMs = ms + } + } + + // Down: read what the server streams until the window closes. + dctx, cancel := context.WithTimeout(ctx, speedWindow) + resp, err := a.speedRequest(dctx, http.MethodGet, "/api/agent/speed/down", nil) + if err != nil { + cancel() + return res, err + } + start := time.Now() + n, _ := io.Copy(io.Discard, resp.Body) + elapsed := time.Since(start) + resp.Body.Close() + cancel() + res.DownBps = bps(n, elapsed) + + // Up: send until the window closes or the cap is hit. + body := &speedSource{deadline: time.Now().Add(speedWindow), left: speedUpCap} + start = time.Now() + resp, err = a.speedRequest(ctx, http.MethodPost, "/api/agent/speed/up", body) + if err != nil { + return res, err + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + res.UpBps = bps(int64(speedUpCap-body.left), time.Since(start)) + return res, nil +} + +func (a *Agent) speedRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, method, a.cfg.Host+path, body) + if err != nil { + return nil, err + } + a.authHeaders(req) + resp, err := a.http.Do(req) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + resp.Body.Close() + return nil, errStatus(resp.StatusCode) + } + return resp, nil +} + +func bps(n int64, d time.Duration) float64 { + if d <= 0 { + return 0 + } + return float64(n) * 8 / d.Seconds() +} + +// speedSource is an upload body of incompressible bytes that ends at the +// deadline or after `left` bytes. +type speedSource struct { + deadline time.Time + left int + block []byte +} + +func (s *speedSource) Read(p []byte) (int, error) { + if s.left <= 0 || !time.Now().Before(s.deadline) { + return 0, io.EOF + } + if s.block == nil { + s.block = make([]byte, 64<<10) + rand.Read(s.block) + } + n := copy(p, s.block) + if n > s.left { + n = s.left + } + s.left -= n + return n, nil +} diff --git a/internal/agent/stat.go b/internal/agent/stat.go new file mode 100644 index 0000000..bc1c416 --- /dev/null +++ b/internal/agent/stat.go @@ -0,0 +1,45 @@ +package agent + +import ( + "runtime" + "sync" + "syscall" + + "github.com/x-grk/longpolling-remote-access/internal/proto" +) + +// statMu serializes samples: the CPU figure is a delta against the last one. +var statMu sync.Mutex + +// sysStat samples the machine's load. It costs nothing while nobody asks: the +// panel requests it only for an open status page. +func sysStat() (proto.SysStatResult, error) { + statMu.Lock() + defer statMu.Unlock() + res := proto.SysStatResult{NCPU: runtime.NumCPU()} + cpu, err := cpuPercent() + if err != nil { + return res, err + } + res.CPU = cpu + if res.MemTotal, res.MemUsed, err = memUsage(); err != nil { + return res, err + } + var fs syscall.Statfs_t + if err := syscall.Statfs("/", &fs); err != nil { + return res, err + } + res.DiskTotal = uint64(fs.Blocks) * uint64(fs.Bsize) + res.DiskFree = uint64(fs.Bavail) * uint64(fs.Bsize) + return res, nil +} + +func clampPercent(v float64) float64 { + if v < 0 { + return 0 + } + if v > 100 { + return 100 + } + return v +} diff --git a/internal/agent/stat_darwin.go b/internal/agent/stat_darwin.go new file mode 100644 index 0000000..b3aa9ce --- /dev/null +++ b/internal/agent/stat_darwin.go @@ -0,0 +1,43 @@ +package agent + +import ( + "bytes" + "os/exec" + "runtime" + "strconv" + + "golang.org/x/sys/unix" +) + +// cpuPercent sums the per-process CPU shares ps reports. macOS exposes the +// host CPU ticks only through Mach calls, which need cgo. +func cpuPercent() (float64, error) { + out, err := exec.Command("/bin/ps", "-A", "-o", "%cpu=").Output() + if err != nil { + return 0, err + } + var sum float64 + for _, f := range bytes.Fields(out) { + // Some locales print a decimal comma. + v, _ := strconv.ParseFloat(string(bytes.Replace(f, []byte(","), []byte("."), 1)), 64) + sum += v + } + return clampPercent(sum / float64(runtime.NumCPU())), nil +} + +// memUsage takes the kernel's own share of memory still available +// (kern.memorystatus_level, what `memory_pressure` prints). +func memUsage() (total, used uint64, err error) { + total, err = unix.SysctlUint64("hw.memsize") + if err != nil { + return 0, 0, err + } + level, err := unix.SysctlUint32("kern.memorystatus_level") + if err != nil { + return 0, 0, err + } + if level > 100 { + level = 100 + } + return total, total / 100 * uint64(100-level), nil +} diff --git a/internal/agent/stat_linux.go b/internal/agent/stat_linux.go new file mode 100644 index 0000000..32b32ec --- /dev/null +++ b/internal/agent/stat_linux.go @@ -0,0 +1,121 @@ +package agent + +import ( + "bytes" + "errors" + "os" + "strconv" + "time" +) + +// Previous /proc/stat reading; the load is the busy share of the ticks since. +var ( + cpuPrevIdle, cpuPrevTotal uint64 + cpuPrevAt time.Time +) + +// cpuStaleAfter is how old the previous reading may be: past it the delta +// would average over a stretch nobody was watching, so take a fresh pair. +const cpuStaleAfter = 15 * time.Second + +func cpuPercent() (float64, error) { + if time.Since(cpuPrevAt) > cpuStaleAfter { + idle, total, err := readCPUTicks() + if err != nil { + return 0, err + } + cpuPrevIdle, cpuPrevTotal = idle, total + time.Sleep(200 * time.Millisecond) + } + idle, total, err := readCPUTicks() + if err != nil { + return 0, err + } + dIdle, dTotal := idle-cpuPrevIdle, total-cpuPrevTotal + cpuPrevIdle, cpuPrevTotal, cpuPrevAt = idle, total, time.Now() + if dTotal == 0 || dIdle > dTotal { + return 0, nil + } + return clampPercent(100 * float64(dTotal-dIdle) / float64(dTotal)), nil +} + +// readCPUTicks parses the aggregate "cpu" line of /proc/stat: +// user nice system idle iowait irq softirq steal ... +func readCPUTicks() (idle, total uint64, err error) { + var buf [512]byte + f, err := os.Open("/proc/stat") + if err != nil { + return 0, 0, err + } + n, _ := f.Read(buf[:]) + f.Close() + line := buf[:n] + if i := bytes.IndexByte(line, '\n'); i >= 0 { + line = line[:i] + } + fields := bytes.Fields(line) + if len(fields) < 5 || string(fields[0]) != "cpu" { + return 0, 0, errors.New("unexpected /proc/stat") + } + for i, f := range fields[1:] { + if i >= 8 { // guest time is already counted in user/nice + break + } + v, _ := strconv.ParseUint(string(f), 10, 64) + total += v + if i == 3 || i == 4 { // idle, iowait + idle += v + } + } + return idle, total, nil +} + +// memUsage reads /proc/meminfo; used is everything not available to new +// processes (so page cache does not count as used). +func memUsage() (total, used uint64, err error) { + data, err := os.ReadFile("/proc/meminfo") + if err != nil { + return 0, 0, err + } + var avail, free uint64 + var haveAvail bool + for len(data) > 0 { + line := data + if i := bytes.IndexByte(data, '\n'); i >= 0 { + line, data = data[:i], data[i+1:] + } else { + data = nil + } + switch { + case bytes.HasPrefix(line, []byte("MemTotal:")): + total = meminfoKB(line) + case bytes.HasPrefix(line, []byte("MemFree:")): + free = meminfoKB(line) + case bytes.HasPrefix(line, []byte("MemAvailable:")): + avail, haveAvail = meminfoKB(line), true + } + if haveAvail && total > 0 { + break + } + } + if total == 0 { + return 0, 0, errors.New("unexpected /proc/meminfo") + } + if !haveAvail { + avail = free + } + if avail > total { + avail = total + } + return total, total - avail, nil +} + +// meminfoKB returns the value of a "Key: 123 kB" line in bytes. +func meminfoKB(line []byte) uint64 { + f := bytes.Fields(line) + if len(f) < 2 { + return 0 + } + v, _ := strconv.ParseUint(string(f[1]), 10, 64) + return v * 1024 +} diff --git a/internal/agent/task.go b/internal/agent/task.go index d5b9de6..c6778c9 100644 --- a/internal/agent/task.go +++ b/internal/agent/task.go @@ -17,9 +17,12 @@ func (a *Agent) runTask(ctx context.Context, taskID string) { return } var res proto.TaskResult - if env.Type == proto.TaskSysUpdate { + switch env.Type { + case proto.TaskSysUpdate: res = a.doUpdate() - } else { + case proto.TaskSysSpeed: + res = a.doSpeed(ctx) + default: res = a.execTask(env) } a.postResult(ctx, taskID, res) diff --git a/internal/agent/users.go b/internal/agent/users.go index c233b81..a634d17 100644 --- a/internal/agent/users.go +++ b/internal/agent/users.go @@ -33,6 +33,8 @@ func dispatchTask(env proto.TaskEnvelope) (any, error) { switch env.Type { case proto.TaskSysUsers: return sysUsers(), nil + case proto.TaskSysStat: + return sysStat() case proto.TaskSysUserAdd: var a proto.SysUserAddArgs _ = json.Unmarshal(env.Args, &a) diff --git a/internal/proto/proto.go b/internal/proto/proto.go index be5567f..cd3008b 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -89,6 +89,8 @@ const ( TaskSysUpdate = "sys.update" // download the latest agent binary and restart TaskSysUsers = "sys.users" // report the agent's user and the login-capable users TaskSysUserAdd = "sys.useradd" // create a new OS user + TaskSysStat = "sys.stat" // CPU load, RAM and disk usage right now + TaskSysSpeed = "sys.speed" // measure the link to the server (ping, down, up) ) // FSReadCap bounds how many bytes TaskFSRead returns (5 MiB). The panel lets a @@ -166,3 +168,21 @@ type SysUserAddArgs struct { Password string `json:"password,omitempty"` // empty leaves the account without a password Sudo bool `json:"sudo"` // add to the sudo/wheel group } + +// SysStatResult is returned for TaskSysStat. Sizes are in bytes. +type SysStatResult struct { + CPU float64 `json:"cpu"` // load across all cores, 0..100 + NCPU int `json:"ncpu"` + MemTotal uint64 `json:"mem_total"` + MemUsed uint64 `json:"mem_used"` + DiskTotal uint64 `json:"disk_total"` // filesystem holding "/" + DiskFree uint64 `json:"disk_free"` +} + +// SysSpeedResult is returned for TaskSysSpeed: the agent's HTTP link to the +// server as the agent measured it. Down is server -> agent. +type SysSpeedResult struct { + PingMs float64 `json:"ping_ms"` + DownBps float64 `json:"down_bps"` // bits per second + UpBps float64 `json:"up_bps"` +} diff --git a/internal/server/agents.go b/internal/server/agents.go index 697ed23..64be65e 100644 --- a/internal/server/agents.go +++ b/internal/server/agents.go @@ -24,6 +24,16 @@ func (s *Server) authenticateAgent(r *http.Request) *Agent { return a } +// online reports whether the agent is connected right now: the hub knows from +// its control polls. Right after a start it cannot tell yet, so the last poll +// recorded in the database decides. +func (s *Server) online(a *Agent) bool { + if on, ok := s.hub.Online(a.ID); ok { + return on + } + return a.recentlySeen() +} + func bearerToken(r *http.Request) string { h := r.Header.Get("Authorization") if strings.HasPrefix(h, "Bearer ") { diff --git a/internal/server/db.go b/internal/server/db.go index aec39db..cc076dd 100644 --- a/internal/server/db.go +++ b/internal/server/db.go @@ -105,6 +105,7 @@ CREATE TABLE IF NOT EXISTS forward_rules ( `ALTER TABLE ufw_rules ADD COLUMN rule_id INTEGER NOT NULL DEFAULT 0`, `ALTER TABLE agents ADD COLUMN run_user TEXT NOT NULL DEFAULT ''`, `ALTER TABLE agents ADD COLUMN login_user TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE forward_rules ADD COLUMN no_ufw INTEGER NOT NULL DEFAULT 0`, } { if _, err := db.sql.Exec(alter); err != nil && !strings.Contains(err.Error(), "duplicate column") { return err @@ -238,8 +239,9 @@ func (a Agent) TerminalUser() string { return firstNonEmpty(a.LoginUser, a.RunUser, "root") } -// Online reports whether the agent polled recently enough to be considered up. -func (a Agent) Online() bool { +// recentlySeen is the fallback for "online" right after a server start, while +// the hub has not seen the agents' polls yet (see Server.online). +func (a Agent) recentlySeen() bool { return time.Since(a.LastSeen) < 45*time.Second } @@ -327,6 +329,7 @@ type ForwardRule struct { HostPort int ListenAddr string Proto string + NoUfw bool // the panel opens no ufw rule for this forward } func (db *DB) CreateForwardRule(r ForwardRule) (int64, error) { @@ -334,9 +337,9 @@ func (db *DB) CreateForwardRule(r ForwardRule) (int64, error) { r.Proto = "http" } res, err := db.sql.Exec( - `INSERT INTO forward_rules(agent_id, client_port, host_port, listen_addr, proto, created_at) - VALUES(?, ?, ?, ?, ?, ?)`, - r.AgentID, r.ClientPort, r.HostPort, r.ListenAddr, r.Proto, time.Now().Unix()) + `INSERT INTO forward_rules(agent_id, client_port, host_port, listen_addr, proto, no_ufw, created_at) + VALUES(?, ?, ?, ?, ?, ?, ?)`, + r.AgentID, r.ClientPort, r.HostPort, r.ListenAddr, r.Proto, r.NoUfw, time.Now().Unix()) if err != nil { return 0, err } @@ -345,7 +348,7 @@ func (db *DB) CreateForwardRule(r ForwardRule) (int64, error) { func (db *DB) ListForwardRules() ([]ForwardRule, error) { rows, err := db.sql.Query( - `SELECT id, agent_id, client_port, host_port, listen_addr, proto FROM forward_rules ORDER BY id`) + `SELECT id, agent_id, client_port, host_port, listen_addr, proto, no_ufw FROM forward_rules ORDER BY id`) if err != nil { return nil, err } @@ -353,7 +356,7 @@ func (db *DB) ListForwardRules() ([]ForwardRule, error) { var out []ForwardRule for rows.Next() { var r ForwardRule - if err := rows.Scan(&r.ID, &r.AgentID, &r.ClientPort, &r.HostPort, &r.ListenAddr, &r.Proto); err != nil { + if err := rows.Scan(&r.ID, &r.AgentID, &r.ClientPort, &r.HostPort, &r.ListenAddr, &r.Proto, &r.NoUfw); err != nil { return nil, err } out = append(out, r) @@ -378,8 +381,8 @@ func (db *DB) ListForwardRulesByAgent(agentID string) ([]ForwardRule, error) { func (db *DB) GetForwardRule(id int64) (*ForwardRule, error) { r := &ForwardRule{} err := db.sql.QueryRow( - `SELECT id, agent_id, client_port, host_port, listen_addr, proto FROM forward_rules WHERE id = ?`, id). - Scan(&r.ID, &r.AgentID, &r.ClientPort, &r.HostPort, &r.ListenAddr, &r.Proto) + `SELECT id, agent_id, client_port, host_port, listen_addr, proto, no_ufw FROM forward_rules WHERE id = ?`, id). + Scan(&r.ID, &r.AgentID, &r.ClientPort, &r.HostPort, &r.ListenAddr, &r.Proto, &r.NoUfw) if err == sql.ErrNoRows { return nil, nil } diff --git a/internal/server/handlers_agent.go b/internal/server/handlers_agent.go index 0258d25..03bfd35 100644 --- a/internal/server/handlers_agent.go +++ b/internal/server/handlers_agent.go @@ -52,7 +52,7 @@ func (s *Server) handlePoll(w http.ResponseWriter, r *http.Request) { } else { _ = s.db.TouchAgent(a.ID) } - msg := s.hub.WaitControl(a.ID, pollHold) + msg := s.hub.WaitControl(r.Context().Done(), a.ID, pollHold) // Set after the hold so a port changed meanwhile reaches the agent now. s.setTunnelHeader(w) writeJSON(w, msg) diff --git a/internal/server/handlers_panel.go b/internal/server/handlers_panel.go index 7619c39..e95e43c 100644 --- a/internal/server/handlers_panel.go +++ b/internal/server/handlers_panel.go @@ -118,7 +118,8 @@ func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { } rows := make([]row, 0, len(agents)) for _, a := range agents { - rows = append(rows, row{Agent: a, Online: a.Online(), Seen: humanSince(a.LastSeen)}) + on := s.online(&a) + rows = append(rows, row{Agent: a, Online: on, Seen: seenText(on, a.LastSeen)}) } s.render(w, "dashboard.html", map[string]any{ "Admin": s.currentAdmin(r), @@ -127,6 +128,36 @@ func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { }) } +// seenText is the "activity" cell: a connected agent is active now, whatever +// the time of its last recorded poll. +func seenText(online bool, lastSeen time.Time) string { + if online { + return "сейчас" + } + return humanSince(lastSeen) +} + +// GET /api/panel/agents/presence — who is online, for the dashboard to keep +// its status column live without reloading. +func (s *Server) handlePresence(w http.ResponseWriter, r *http.Request) { + agents, err := s.db.ListAgents() + if err != nil { + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + type item struct { + ID string `json:"id"` + Online bool `json:"online"` + Seen string `json:"seen"` + } + out := make([]item, 0, len(agents)) + for _, a := range agents { + on := s.online(&a) + out = append(out, item{ID: a.ID, Online: on, Seen: seenText(on, a.LastSeen)}) + } + writeJSON(w, out) +} + func (s *Server) handleTerminalPage(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") a, err := s.db.GetAgent(id) @@ -153,6 +184,8 @@ func (s *Server) handleAgentDelete(w http.ResponseWriter, r *http.Request) { s.removeForward(rule.ID) } _ = s.db.DeleteAgent(id) + s.stats.drop(id) + s.hub.Forget(id) http.Redirect(w, r, "/", http.StatusSeeOther) } @@ -173,7 +206,7 @@ func (s *Server) handleAgentEdit(w http.ResponseWriter, r *http.Request) { "UserErr": r.URL.Query().Get("uerr"), } // The user list lives on the machine, so it is only known while it is online. - if a.Online() { + if s.online(a) { if users, err := s.agentUsers(id); err == nil { data["Users"] = users } else { @@ -224,7 +257,7 @@ func (s *Server) handleAgentSetUser(w http.ResponseWriter, r *http.Request) { return } name := strings.TrimSpace(r.FormValue("user")) - if !a.Online() { + if !s.online(a) { editUserRedirect(w, r, id, "uerr", "Машина офлайн — сменить пользователя нельзя.") return } @@ -265,7 +298,7 @@ func (s *Server) handleAgentAddUser(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } - if !a.Online() { + if !s.online(a) { editUserRedirect(w, r, id, "uerr", "Машина офлайн — создать пользователя нельзя.") return } @@ -306,7 +339,7 @@ func (s *Server) handleAgentUpdate(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } - if !a.Online() { + if !s.online(a) { http.Redirect(w, r, "/agents/"+id+"/edit?updated=0", http.StatusSeeOther) return } diff --git a/internal/server/handlers_panel_fs.go b/internal/server/handlers_panel_fs.go index 20819ea..736a0ec 100644 --- a/internal/server/handlers_panel_fs.go +++ b/internal/server/handlers_panel_fs.go @@ -53,7 +53,7 @@ func (s *Server) handleFSOp(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } - if !a.Online() { + if !s.online(a) { writeJSON(w, proto.TaskResult{Error: "машина офлайн"}) return } @@ -83,7 +83,7 @@ func (s *Server) handleFSDownload(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } - if !a.Online() { + if !s.online(a) { http.Error(w, "машина офлайн", http.StatusServiceUnavailable) return } diff --git a/internal/server/handlers_panel_fwd.go b/internal/server/handlers_panel_fwd.go index 9c159d2..778ff7a 100644 --- a/internal/server/handlers_panel_fwd.go +++ b/internal/server/handlers_panel_fwd.go @@ -30,6 +30,7 @@ func (s *Server) handleForwardsPage(w http.ResponseWriter, r *http.Request) { "Error": errMsg, "TunnelPort": s.TunnelPort(), "PanelPort": panelPort(s.db.HostAddress()), + "UfwEnabled": s.ufw.enabled(), }) } @@ -85,7 +86,8 @@ func (s *Server) handleForwardAdd(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, "/agents/"+id+"/forwards?err=busy", http.StatusSeeOther) return } - rule := ForwardRule{AgentID: id, ClientPort: clientPort, HostPort: hostPort, ListenAddr: listenAddr, Proto: protoKind} + rule := ForwardRule{AgentID: id, ClientPort: clientPort, HostPort: hostPort, ListenAddr: listenAddr, Proto: protoKind, + NoUfw: r.FormValue("no_ufw") != ""} ruleID, err := s.db.CreateForwardRule(rule) if err != nil { http.Redirect(w, r, "/agents/"+id+"/forwards?err=db", http.StatusSeeOther) diff --git a/internal/server/handlers_panel_stat.go b/internal/server/handlers_panel_stat.go new file mode 100644 index 0000000..73c90e9 --- /dev/null +++ b/internal/server/handlers_panel_stat.go @@ -0,0 +1,193 @@ +package server + +import ( + "crypto/rand" + "errors" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/x-grk/longpolling-remote-access/internal/proto" +) + +const ( + // statTaskTimeout bounds how long a status request waits for the agent. + statTaskTimeout = 8 * time.Second + // statCacheFor is how long one sample answers every open status page, so + // several tabs cost the agent a single reading. + statCacheFor = 2 * time.Second + // speedTaskTimeout covers the agent's ping + down + up windows with slack. + speedTaskTimeout = 30 * time.Second + // speedDownCap bounds one download stream; speedDownFor ends it if the + // agent does not hang up first. + speedDownCap = 256 << 20 + speedDownFor = 6 * time.Second + speedUpCap = 64 << 20 +) + +// statCache holds the last sample per agent and serializes its refresh. +type statCache struct { + mu sync.Mutex + m map[string]*statEntry +} + +type statEntry struct { + mu sync.Mutex + at time.Time + res proto.TaskResult +} + +func (c *statCache) entry(agentID string) *statEntry { + c.mu.Lock() + defer c.mu.Unlock() + if c.m == nil { + c.m = map[string]*statEntry{} + } + e := c.m[agentID] + if e == nil { + e = &statEntry{} + c.m[agentID] = e + } + return e +} + +func (c *statCache) drop(agentID string) { + c.mu.Lock() + delete(c.m, agentID) + c.mu.Unlock() +} + +func (s *Server) handleStatusPage(w http.ResponseWriter, r *http.Request) { + a, err := s.db.GetAgent(r.PathValue("id")) + if err != nil || a == nil { + http.NotFound(w, r) + return + } + s.render(w, "status.html", map[string]any{"Agent": a}) +} + +// runTask runs a one-shot task on the agent and waits for its result. +func (s *Server) runTask(agentID, typ string, timeout time.Duration) (proto.TaskResult, error) { + taskID, ch := s.hub.CreateTask(agentID, typ, nil) + defer s.hub.DiscardTask(taskID) + t := time.NewTimer(timeout) + defer t.Stop() + select { + case res := <-ch: + return res, nil + case <-t.C: + return proto.TaskResult{}, errors.New("таймаут ожидания клиента") + } +} + +// onlineAgent answers the request itself and returns false when the agent is +// unknown or offline. +func (s *Server) onlineAgent(w http.ResponseWriter, r *http.Request) (string, bool) { + id := r.PathValue("id") + a, err := s.db.GetAgent(id) + if err != nil || a == nil { + http.NotFound(w, r) + return "", false + } + if !s.online(a) { + writeJSON(w, proto.TaskResult{Error: "машина офлайн"}) + return "", false + } + return id, true +} + +// oldAgentError replaces the "unknown task" answer of an agent that predates +// the task. +func oldAgentError(res proto.TaskResult) proto.TaskResult { + if !res.OK && strings.Contains(res.Error, "неизвестная операция") { + res.Error = "клиент не поддерживает эту функцию — обновите его (нужен 0.0.5+)" + } + return res +} + +// GET /api/panel/agents/{id}/stat — CPU, RAM and disk of the machine right now. +func (s *Server) handleAgentStat(w http.ResponseWriter, r *http.Request) { + id, ok := s.onlineAgent(w, r) + if !ok { + return + } + e := s.stats.entry(id) + e.mu.Lock() + defer e.mu.Unlock() + if time.Since(e.at) > statCacheFor { + res, err := s.runTask(id, proto.TaskSysStat, statTaskTimeout) + if err != nil { + // Cached too, so pages queued behind this one don't each wait it out. + res = proto.TaskResult{Error: err.Error()} + } + e.res, e.at = oldAgentError(res), time.Now() + } + writeJSON(w, e.res) +} + +// POST /api/panel/agents/{id}/speed — have the agent measure its link to us. +func (s *Server) handleAgentSpeed(w http.ResponseWriter, r *http.Request) { + id, ok := s.onlineAgent(w, r) + if !ok { + return + } + res, err := s.runTask(id, proto.TaskSysSpeed, speedTaskTimeout) + if err != nil { + writeJSON(w, proto.TaskResult{Error: err.Error()}) + return + } + writeJSON(w, oldAgentError(res)) +} + +// speedBlock is one chunk of incompressible bytes, made once and sent over and +// over, so a speed test costs the server no CPU beyond the copy. +var speedBlock = sync.OnceValue(func() []byte { + b := make([]byte, 64<<10) + rand.Read(b) + return b +}) + +// GET /api/agent/speed/down?bytes=N — stream N bytes (default: as much as fits +// in the time window) for the agent to time. bytes=0 is the ping probe. +func (s *Server) handleSpeedDown(w http.ResponseWriter, r *http.Request) { + if s.authenticateAgent(r) == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + left := speedDownCap + if v := r.URL.Query().Get("bytes"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n >= 0 && n < left { + left = n + } + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Cache-Control", "no-store") + // Keep a proxy in front from buffering or compressing the stream. + w.Header().Set("X-Accel-Buffering", "no") + block := speedBlock() + deadline := time.Now().Add(speedDownFor) + for left > 0 && time.Now().Before(deadline) && r.Context().Err() == nil { + chunk := block + if len(chunk) > left { + chunk = chunk[:left] + } + n, err := w.Write(chunk) + if err != nil { + return + } + left -= n + } +} + +// POST /api/agent/speed/up — swallow the agent's upload. +func (s *Server) handleSpeedUp(w http.ResponseWriter, r *http.Request) { + if s.authenticateAgent(r) == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + _, _ = io.Copy(io.Discard, http.MaxBytesReader(w, r.Body, speedUpCap)) + w.WriteHeader(http.StatusNoContent) +} diff --git a/internal/server/hub.go b/internal/server/hub.go index ff4c7cf..c0f1be5 100644 --- a/internal/server/hub.go +++ b/internal/server/hub.go @@ -22,6 +22,12 @@ type Hub struct { sessions map[string]*Session tasks map[string]*taskEntry + + // Presence: control polls held open right now per agent, and when each + // agent's last one ended. started is when this hub came up. + polls map[string]int + lastPoll map[string]time.Time + started time.Time } // taskEntry is a pending one-shot task awaiting the agent's result. @@ -52,6 +58,9 @@ func NewHub() *Hub { ctrlWaiters: map[string]chan struct{}{}, sessions: map[string]*Session{}, tasks: map[string]*taskEntry{}, + polls: map[string]int{}, + lastPoll: map[string]time.Time{}, + started: time.Now(), } } @@ -97,9 +106,56 @@ func (h *Hub) CreateForward(agentID string, port int) *Session { return s } -// WaitControl blocks until a job is queued for agentID or timeout elapses. -func (h *Hub) WaitControl(agentID string, timeout time.Duration) proto.ControlMessage { - deadline := time.After(timeout) +// pollGrace is how long an agent stays online between two control polls: it +// re-polls at once, so this only has to cover a slow round trip or one retry. +const pollGrace = 10 * time.Second + +// Online reports whether the agent is connected: it holds a control poll open +// or has just finished one. ok is false while that cannot be told yet (right +// after a start, before agents had time to come back). +func (h *Hub) Online(agentID string) (online, ok bool) { + h.mu.Lock() + defer h.mu.Unlock() + if h.polls[agentID] > 0 { + return true, true + } + if t, seen := h.lastPoll[agentID]; seen { + return time.Since(t) < pollGrace, true + } + return false, time.Since(h.started) > pollHold+pollGrace +} + +// Forget drops what the hub keeps about a deleted agent. +func (h *Hub) Forget(agentID string) { + h.mu.Lock() + delete(h.jobs, agentID) + delete(h.ctrlWaiters, agentID) + delete(h.lastPoll, agentID) + h.mu.Unlock() +} + +// WaitControl blocks until a job is queued for agentID, the timeout elapses +// or the agent hangs up (done). The agent counts as online meanwhile. +func (h *Hub) WaitControl(done <-chan struct{}, agentID string, timeout time.Duration) proto.ControlMessage { + h.mu.Lock() + h.polls[agentID]++ + h.mu.Unlock() + defer func() { + h.mu.Lock() + if h.polls[agentID]--; h.polls[agentID] <= 0 { + delete(h.polls, agentID) + } + // A poll cut off by the agent is not a sign of life. + select { + case <-done: + default: + h.lastPoll[agentID] = time.Now() + } + h.mu.Unlock() + }() + t := time.NewTimer(timeout) + defer t.Stop() + deadline := t.C for { h.mu.Lock() if q := h.jobs[agentID]; len(q) > 0 { @@ -119,6 +175,9 @@ func (h *Hub) WaitControl(agentID string, timeout time.Duration) proto.ControlMe case <-w: case <-deadline: return proto.ControlMessage{Type: proto.CtrlIdle} + case <-done: + // The job, if one comes, stays queued for the next poll. + return proto.ControlMessage{Type: proto.CtrlIdle} } } } diff --git a/internal/server/server.go b/internal/server/server.go index 0018527..ddfe75a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -52,6 +52,8 @@ type Server struct { tunnelDefault int tunnelMu sync.Mutex // serializes port changes tunnelPort atomic.Int64 + + stats statCache // last load sample per agent, for the status page } // New builds the server, parsing embedded templates. @@ -200,6 +202,7 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /agents/{id}/terminal", s.requireAdmin(s.handleTerminalPage)) mux.Handle("GET /agents/{id}/files", s.requireAdmin(s.handleFilesPage)) mux.Handle("GET /agents/{id}/forwards", s.requireAdmin(s.handleForwardsPage)) + mux.Handle("GET /agents/{id}/status", s.requireAdmin(s.handleStatusPage)) mux.Handle("GET /agents/{id}/edit", s.requireAdmin(s.handleAgentEdit)) mux.Handle("POST /agents/{id}/rename", s.requireAdmin(s.handleAgentRename)) mux.Handle("POST /agents/{id}/update", s.requireAdmin(s.handleAgentUpdate)) @@ -225,6 +228,11 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/panel/agents/{id}/fs/download", s.requireAdmin(s.handleFSDownload)) mux.Handle("POST /api/panel/agents/{id}/fs/{op}", s.requireAdmin(s.handleFSOp)) + // Machine status (admin). + mux.Handle("GET /api/panel/agents/presence", s.requireAdmin(s.handlePresence)) + mux.Handle("GET /api/panel/agents/{id}/stat", s.requireAdmin(s.handleAgentStat)) + mux.Handle("POST /api/panel/agents/{id}/speed", s.requireAdmin(s.handleAgentSpeed)) + // Panel port forwarding (admin-only). mux.Handle("POST /agents/{id}/forwards/add", s.requireAdmin(s.handleForwardAdd)) mux.Handle("POST /forwards/{rid}/delete", s.requireAdmin(s.handleForwardDelete)) @@ -237,6 +245,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/agent/term/{sid}/out", s.handleTermOut) mux.HandleFunc("GET /api/agent/task/{id}", s.handleTaskFetch) mux.HandleFunc("POST /api/agent/task/{id}/result", s.handleTaskResult) + mux.HandleFunc("GET /api/agent/speed/down", s.handleSpeedDown) + mux.HandleFunc("POST /api/agent/speed/up", s.handleSpeedUp) return s.logRequests(mux) } diff --git a/internal/server/ufw.go b/internal/server/ufw.go index ddc960a..9619fab 100644 --- a/internal/server/ufw.go +++ b/internal/server/ufw.go @@ -54,6 +54,16 @@ func (u *ufwManager) status() (installed, active, runnable bool) { return installed, active, runnable } +// enabled reports whether the integration is switched on and ufw is there to +// drive. Unlike available it runs nothing, so pages can ask it freely. +func (u *ufwManager) enabled() bool { + if !u.db.UfwEnabled() { + return false + } + _, err := exec.LookPath("ufw") + return err == nil +} + // available reports whether we can and should drive ufw now. func (u *ufwManager) available() bool { if !u.db.UfwEnabled() { @@ -148,7 +158,9 @@ func (u *ufwManager) pruneShared() { } used := map[string]bool{} for _, f := range forwards { - used[f.Proto] = true + if !f.NoUfw { + used[f.Proto] = true + } } port := u.tunnelPort() for _, r := range shared { @@ -160,7 +172,8 @@ func (u *ufwManager) pruneShared() { // allowForward opens the host ports a forward rule needs: its host port (bound // to the rule so it is revoked with it) plus the shared tunnel port for direct -// tcp/udp modes (bound to 0 — shared across rules). +// tcp/udp modes (bound to 0 — shared across rules). A rule made "without ufw" +// opens nothing, so nothing is revoked with it either. func (u *ufwManager) allowForward(rule ForwardRule) { if !u.available() { return @@ -169,6 +182,9 @@ func (u *ufwManager) allowForward(rule ForwardRule) { } func (u *ufwManager) allowForwardPorts(rule ForwardRule) { + if rule.NoUfw { + return + } // "http" (long-poll) and "tcp" forwards both listen on TCP on the host. hostProto, direct := "tcp", rule.Proto == "tcp" if rule.Proto == "udp" { diff --git a/internal/server/web/templates/dashboard.html b/internal/server/web/templates/dashboard.html index b8af8b9..e8f7322 100644 --- a/internal/server/web/templates/dashboard.html +++ b/internal/server/web/templates/dashboard.html @@ -25,17 +25,26 @@ СтатусИмяHostnameОС / АрхАктивность {{range .Agents}} - - {{if .Online}}online{{else}}offline{{end}} + + {{if .Online}}online{{else}}offline{{end}} {{.Name}} {{.Hostname}} {{.OS}}/{{.Arch}} - {{.Seen}} + {{.Seen}}
- Терминал - Файлы - Порты + + + + + + + + + + + + @@ -58,5 +67,38 @@ {{end}}
+ diff --git a/internal/server/web/templates/forwards.html b/internal/server/web/templates/forwards.html index 4cbf1cf..5bdb936 100644 --- a/internal/server/web/templates/forwards.html +++ b/internal/server/web/templates/forwards.html @@ -12,6 +12,7 @@ .frow select, .frow input, .frow button { height:40px; box-sizing:border-box; font-size:14px; line-height:1; margin:0; } + label.check { display:inline-flex; align-items:center; gap:8px; margin:14px 0 0; color:var(--text); font-size:14px; cursor:pointer; } .frow select { padding:0 32px 0 12px; border-radius:8px; border:1px solid var(--border); background:#0c0f13; color:var(--text); @@ -44,7 +45,7 @@ {{.Proto}} :{{.HostPort}} 127.0.0.1:{{.ClientPort}} - {{.ListenAddr}} + {{.ListenAddr}}{{if and .NoUfw $.UfwEnabled}} · без ufw{{end}}
@@ -88,6 +89,11 @@ + {{if .UfwEnabled}} + + {{end}}

diff --git a/internal/server/web/templates/status.html b/internal/server/web/templates/status.html new file mode 100644 index 0000000..18c5a9f --- /dev/null +++ b/internal/server/web/templates/status.html @@ -0,0 +1,156 @@ + + + + + +Статус · {{.Agent.Name}} — LPRA + + + + +
+
LPRA · Статус — {{.Agent.Name}} ({{.Agent.Hostname}})
+ +
+
+

Статус

+ + +
+
+
Процессор—
+
+
+
+
+
Оперативная память—
+
+
+
+
+
Диск (/)—
+
+
+
+
+ +
+

Скорость соединения

+

Клиент замеряет свой канал до хоста панели — тот же, по которому идут терминал, файлы и http-проброс. + Замер занимает около 7 секунд и на это время нагружает канал.

+
+ + +
+ +
+
+ + +