2603 lines
75 KiB
Go
2603 lines
75 KiB
Go
package cluster
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ed25519"
|
|
"crypto/rand"
|
|
"crypto/sha256"
|
|
"crypto/x509"
|
|
"crypto/x509/pkix"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"encoding/pem"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math/big"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"os/user"
|
|
"path"
|
|
"path/filepath"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"golang.org/x/crypto/bcrypt"
|
|
"golang.org/x/crypto/ssh"
|
|
"golang.org/x/crypto/ssh/knownhosts"
|
|
|
|
"pxmon/internal/agent"
|
|
"pxmon/internal/history"
|
|
)
|
|
|
|
const (
|
|
defaultSSHPort = 22
|
|
defaultProbeTO = 8 * time.Second
|
|
defaultAgentPort = 19090
|
|
defaultAgentListen = "0.0.0.0:19090"
|
|
)
|
|
|
|
var (
|
|
ErrClusterNotFound = errors.New("cluster not found")
|
|
ErrNoActiveCluster = errors.New("no active cluster selected")
|
|
)
|
|
|
|
// Service manages node inventory and SSH/agent operations.
|
|
type Service struct {
|
|
store *Store
|
|
now func() time.Time
|
|
|
|
tunnelMu sync.Mutex
|
|
tunnelPool map[string]*tunneledSSH
|
|
|
|
networkStore *history.NetworkStore
|
|
}
|
|
|
|
// AttachNetworkStore wires a persistent network history store to the service
|
|
// so non-TUI callers (bot, CLI one-shots) can read the same data the TUI
|
|
// monitor writes.
|
|
func (s *Service) AttachNetworkStore(store *history.NetworkStore) {
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.networkStore = store
|
|
}
|
|
|
|
// NetworkStore returns the attached network history store or nil.
|
|
func (s *Service) NetworkStore() *history.NetworkStore {
|
|
if s == nil {
|
|
return nil
|
|
}
|
|
return s.networkStore
|
|
}
|
|
|
|
// tunneledSSH caches an ssh.Client used to tunnel agent HTTP traffic for an
|
|
// ipfabric node. Reused across probes so we don't pay a new SSH handshake per
|
|
// HTTP call.
|
|
type tunneledSSH struct {
|
|
client *ssh.Client
|
|
fp string // credential fingerprint; lets us invalidate on re-auth
|
|
}
|
|
|
|
func NewService(store *Store) *Service {
|
|
return &Service{
|
|
store: store,
|
|
now: time.Now,
|
|
tunnelPool: make(map[string]*tunneledSSH),
|
|
}
|
|
}
|
|
|
|
func (s *Service) DataDir() string {
|
|
if s == nil || s.store == nil {
|
|
return "."
|
|
}
|
|
return filepath.Dir(s.store.Path())
|
|
}
|
|
|
|
func (s *Service) ConfigPath() string {
|
|
if s == nil || s.store == nil {
|
|
return ""
|
|
}
|
|
return s.store.Path()
|
|
}
|
|
|
|
// ConnectOptions contains SSH node registration parameters.
|
|
type ConnectOptions struct {
|
|
Name string
|
|
Host string
|
|
Port int
|
|
User string
|
|
Transport TransportMode
|
|
AuthMethod AuthMethod
|
|
Password string
|
|
StorePassword bool
|
|
KeyPath string
|
|
KeyPassphrase string
|
|
KeyPassphraseFile string
|
|
StoreKeyPassphrase bool
|
|
StoreKeyPassphraseFile bool
|
|
InsecureHostKey bool
|
|
SkipCheck bool
|
|
AllowUnreachable bool
|
|
Force bool
|
|
}
|
|
|
|
// SSHProbeResult describes SSH connectivity status.
|
|
type SSHProbeResult struct {
|
|
Reachable bool `json:"reachable"`
|
|
Address string `json:"address"`
|
|
LatencyMS int64 `json:"latency_ms"`
|
|
Error string `json:"error,omitempty"`
|
|
CheckedAt time.Time `json:"checked_at"`
|
|
}
|
|
|
|
// AgentPingResult describes pxmon-agent availability.
|
|
type AgentPingResult struct {
|
|
Reachable bool `json:"reachable"`
|
|
Endpoint string `json:"endpoint"`
|
|
StatusCode int `json:"status_code"`
|
|
Version string `json:"version,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
CheckedAt time.Time `json:"checked_at"`
|
|
}
|
|
|
|
// BootstrapOptions configures agent installation over SSH.
|
|
type BootstrapOptions struct {
|
|
Selector string
|
|
Password string
|
|
KeyPassphrase string
|
|
ListenAddress string
|
|
AgentPort int
|
|
LocalAgentBin string
|
|
RotateToken bool
|
|
AllowAgentProbe bool
|
|
}
|
|
|
|
// BootstrapResult returns details about deployed agent.
|
|
type BootstrapResult struct {
|
|
RemoteOS string `json:"remote_os"`
|
|
RemoteArch string `json:"remote_arch"`
|
|
PID string `json:"pid"`
|
|
AgentPing AgentPingResult `json:"agent_ping"`
|
|
}
|
|
|
|
func (s *Service) Connect(ctx context.Context, opts ConnectOptions) (Cluster, SSHProbeResult, error) {
|
|
name := strings.TrimSpace(opts.Name)
|
|
host := strings.TrimSpace(opts.Host)
|
|
if name == "" {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--name is required")
|
|
}
|
|
if host == "" {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--host is required")
|
|
}
|
|
if strings.Contains(host, "://") {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--host must be hostname or IP, not URL")
|
|
}
|
|
|
|
port := opts.Port
|
|
if port == 0 {
|
|
port = defaultSSHPort
|
|
}
|
|
if port < 1 || port > 65535 {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--port must be in range 1..65535")
|
|
}
|
|
|
|
username := strings.TrimSpace(opts.User)
|
|
if username == "" {
|
|
if u, err := user.Current(); err == nil {
|
|
username = u.Username
|
|
}
|
|
}
|
|
if username == "" {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--user is required")
|
|
}
|
|
|
|
authMethod := opts.AuthMethod
|
|
if authMethod == "" {
|
|
authMethod = AuthMethodKey
|
|
}
|
|
if authMethod != AuthMethodPassword && authMethod != AuthMethodKey {
|
|
return Cluster{}, SSHProbeResult{}, fmt.Errorf("unsupported auth method %q", authMethod)
|
|
}
|
|
|
|
keyPath := ""
|
|
if authMethod == AuthMethodKey {
|
|
if strings.TrimSpace(opts.KeyPath) == "" {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--key-path is required for key auth")
|
|
}
|
|
expanded, err := expandPath(opts.KeyPath)
|
|
if err != nil {
|
|
return Cluster{}, SSHProbeResult{}, err
|
|
}
|
|
if _, err := os.Stat(expanded); err != nil {
|
|
return Cluster{}, SSHProbeResult{}, fmt.Errorf("read key file: %w", err)
|
|
}
|
|
keyPath = expanded
|
|
}
|
|
if authMethod == AuthMethodPassword && strings.TrimSpace(opts.Password) == "" {
|
|
return Cluster{}, SSHProbeResult{}, errors.New("--password is required for password auth")
|
|
}
|
|
|
|
storedPassword := ""
|
|
if opts.StorePassword {
|
|
storedPassword = opts.Password
|
|
}
|
|
storedPassphrase := ""
|
|
if opts.StoreKeyPassphrase {
|
|
storedPassphrase = opts.KeyPassphrase
|
|
}
|
|
storedPassphraseFile := ""
|
|
if opts.StoreKeyPassphraseFile {
|
|
expanded, err := expandPath(opts.KeyPassphraseFile)
|
|
if err != nil {
|
|
return Cluster{}, SSHProbeResult{}, err
|
|
}
|
|
if _, err := os.Stat(expanded); err != nil {
|
|
return Cluster{}, SSHProbeResult{}, fmt.Errorf("read key passphrase file: %w", err)
|
|
}
|
|
storedPassphraseFile = expanded
|
|
}
|
|
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, SSHProbeResult{}, err
|
|
}
|
|
|
|
now := s.now().UTC()
|
|
idxByName := -1
|
|
for i, c := range reg.Clusters {
|
|
if strings.EqualFold(c.Name, name) {
|
|
idxByName = i
|
|
break
|
|
}
|
|
}
|
|
|
|
candidate := Cluster{
|
|
ID: newClusterID(),
|
|
Name: name,
|
|
Host: host,
|
|
Port: port,
|
|
User: username,
|
|
Transport: normalizeTransport(opts.Transport),
|
|
AuthMethod: authMethod,
|
|
Password: storedPassword,
|
|
KeyPath: keyPath,
|
|
KeyPassphrase: storedPassphrase,
|
|
KeyPassphraseFile: storedPassphraseFile,
|
|
InsecureHostKey: opts.InsecureHostKey,
|
|
Alerts: defaultAlertPolicy(),
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
|
|
var probe SSHProbeResult
|
|
if !opts.SkipCheck {
|
|
probe = s.probeSSH(ctx, candidate, opts.Password, opts.KeyPassphrase)
|
|
if !probe.Reachable && !opts.AllowUnreachable {
|
|
return Cluster{}, probe, fmt.Errorf("ssh probe failed, use --allow-unreachable to save anyway: %s", probe.Error)
|
|
}
|
|
if probe.Reachable {
|
|
if software, softErr := s.probeSoftware(ctx, candidate, opts.Password, opts.KeyPassphrase); softErr == nil {
|
|
candidate.Software = software
|
|
}
|
|
}
|
|
}
|
|
|
|
if idxByName >= 0 {
|
|
if !opts.Force {
|
|
return Cluster{}, probe, fmt.Errorf("cluster with name %q already exists, use --force to overwrite", name)
|
|
}
|
|
|
|
existing := reg.Clusters[idxByName]
|
|
existing.Host = candidate.Host
|
|
existing.Port = candidate.Port
|
|
existing.User = candidate.User
|
|
existing.Transport = candidate.Transport
|
|
existing.AuthMethod = candidate.AuthMethod
|
|
existing.Password = candidate.Password
|
|
existing.KeyPath = candidate.KeyPath
|
|
existing.KeyPassphrase = candidate.KeyPassphrase
|
|
existing.KeyPassphraseFile = candidate.KeyPassphraseFile
|
|
existing.InsecureHostKey = candidate.InsecureHostKey
|
|
existing.Alerts = ensureAlertPolicy(existing.Alerts)
|
|
if !candidate.Software.DetectedAt.IsZero() {
|
|
existing.Software = candidate.Software
|
|
}
|
|
existing.UpdatedAt = now
|
|
|
|
reg.Clusters[idxByName] = existing
|
|
reg.ActiveClusterID = existing.ID
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, probe, err
|
|
}
|
|
return existing, probe, nil
|
|
}
|
|
|
|
if existsTarget(reg.Clusters, candidate.Host, candidate.Port, candidate.User) {
|
|
return Cluster{}, probe, fmt.Errorf("cluster with target %s@%s:%d already exists", candidate.User, candidate.Host, candidate.Port)
|
|
}
|
|
|
|
reg.Clusters = append(reg.Clusters, candidate)
|
|
reg.ActiveClusterID = candidate.ID
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, probe, err
|
|
}
|
|
|
|
return candidate, probe, nil
|
|
}
|
|
|
|
func (s *Service) GetAlertPolicy(selector string) (AlertPolicy, error) {
|
|
c, err := s.Get(selector)
|
|
if err != nil {
|
|
return AlertPolicy{}, err
|
|
}
|
|
return ensureAlertPolicy(c.Alerts), nil
|
|
}
|
|
|
|
func (s *Service) SetAlertPolicy(selector string, policy AlertPolicy) (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
cluster, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
cluster.Alerts = ensureAlertPolicy(policy)
|
|
cluster.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = cluster
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
return cluster, nil
|
|
}
|
|
|
|
func (s *Service) GetTelegram() (Telegram, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
return normalizeTelegram(reg.Telegram), nil
|
|
}
|
|
|
|
func (s *Service) SetTelegram(cfg Telegram) (Telegram, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
|
|
normalized, err := validateTelegramInput(cfg)
|
|
if err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
normalized.UpdatedAt = s.now().UTC()
|
|
reg.Telegram = normalized
|
|
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
return normalized, nil
|
|
}
|
|
|
|
func (s *Service) DisableTelegram() (Telegram, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
|
|
cfg := normalizeTelegram(reg.Telegram)
|
|
cfg.Enabled = false
|
|
cfg.UpdatedAt = s.now().UTC()
|
|
reg.Telegram = cfg
|
|
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Telegram{}, err
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
func (s *Service) GetLocker() (Locker, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Locker{}, err
|
|
}
|
|
return normalizeLocker(reg.Locker), nil
|
|
}
|
|
|
|
func (s *Service) AuditLocker(event, detail string) {
|
|
if s == nil || s.store == nil {
|
|
return
|
|
}
|
|
_ = s.store.AppendLockerAudit(event, detail)
|
|
}
|
|
|
|
func (s *Service) SetLockerPassword(password string) (Locker, error) {
|
|
pass := strings.TrimSpace(password)
|
|
if len(pass) < 4 {
|
|
s.AuditLocker("locker_set_password_failed", "password too short")
|
|
return Locker{}, errors.New("locker password must be at least 4 characters")
|
|
}
|
|
hash, err := bcrypt.GenerateFromPassword([]byte(pass), bcrypt.DefaultCost)
|
|
if err != nil {
|
|
s.AuditLocker("locker_set_password_failed", "bcrypt error")
|
|
return Locker{}, err
|
|
}
|
|
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Locker{}, err
|
|
}
|
|
cfg := normalizeLocker(reg.Locker)
|
|
cfg.PasswordHash = string(hash)
|
|
cfg.Enabled = true
|
|
cfg.UpdatedAt = s.now().UTC()
|
|
reg.Locker = cfg
|
|
if err := s.store.Save(reg); err != nil {
|
|
s.AuditLocker("locker_set_password_failed", "save registry failed")
|
|
return Locker{}, err
|
|
}
|
|
_ = s.store.ClearLockerSession()
|
|
if err := s.store.SaveLockerSession(cfg.PasswordHash, 6*time.Hour); err != nil {
|
|
s.AuditLocker("locker_set_password_failed", "save locker session failed")
|
|
return Locker{}, err
|
|
}
|
|
s.AuditLocker("locker_set_password", "locker enabled and session issued")
|
|
return cfg, nil
|
|
}
|
|
|
|
func (s *Service) SetLockerEnabled(enabled bool) (Locker, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Locker{}, err
|
|
}
|
|
cfg := normalizeLocker(reg.Locker)
|
|
if enabled && strings.TrimSpace(cfg.PasswordHash) == "" {
|
|
s.AuditLocker("locker_set_enabled_failed", "password not set")
|
|
return Locker{}, errors.New("locker password is not set")
|
|
}
|
|
cfg.Enabled = enabled
|
|
cfg.UpdatedAt = s.now().UTC()
|
|
reg.Locker = cfg
|
|
if err := s.store.Save(reg); err != nil {
|
|
s.AuditLocker("locker_set_enabled_failed", "save registry failed")
|
|
return Locker{}, err
|
|
}
|
|
if !enabled {
|
|
_ = s.store.ClearLockerSession()
|
|
s.AuditLocker("locker_disabled", "session cleared")
|
|
} else {
|
|
s.AuditLocker("locker_enabled", "locker enabled")
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
func (s *Service) UnlockLocker(password string) error {
|
|
cfg, err := s.GetLocker()
|
|
if err != nil {
|
|
s.AuditLocker("locker_unlock_failed", "load locker config failed")
|
|
return err
|
|
}
|
|
if !cfg.Enabled {
|
|
return nil
|
|
}
|
|
if strings.TrimSpace(cfg.PasswordHash) == "" {
|
|
s.AuditLocker("locker_unlock_failed", "password hash not set")
|
|
return errors.New("locker password is not set")
|
|
}
|
|
if err := bcrypt.CompareHashAndPassword([]byte(cfg.PasswordHash), []byte(password)); err != nil {
|
|
s.AuditLocker("locker_unlock_failed", "invalid password")
|
|
return errors.New("invalid locker password")
|
|
}
|
|
if err := s.store.SaveLockerSession(cfg.PasswordHash, 6*time.Hour); err != nil {
|
|
s.AuditLocker("locker_unlock_failed", "save session failed")
|
|
return err
|
|
}
|
|
s.AuditLocker("locker_unlocked", "session issued for 6h")
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) LockNow() error {
|
|
if err := s.store.ClearLockerSession(); err != nil {
|
|
s.AuditLocker("locker_lock_now_failed", "clear session failed")
|
|
return err
|
|
}
|
|
s.AuditLocker("locker_locked", "session cleared")
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) IsLocked() (bool, time.Time, error) {
|
|
cfg, err := s.GetLocker()
|
|
if err != nil {
|
|
return true, time.Time{}, err
|
|
}
|
|
if !cfg.Enabled {
|
|
return false, time.Time{}, nil
|
|
}
|
|
ok, expiresAt, err := s.store.ValidateLockerSession(cfg.PasswordHash)
|
|
if err != nil {
|
|
return true, time.Time{}, err
|
|
}
|
|
return !ok, expiresAt, nil
|
|
}
|
|
|
|
func (s *Service) List() ([]Cluster, string, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
|
|
clusters := append([]Cluster(nil), reg.Clusters...)
|
|
sort.Slice(clusters, func(i, j int) bool {
|
|
return strings.ToLower(clusters[i].Name) < strings.ToLower(clusters[j].Name)
|
|
})
|
|
|
|
return clusters, reg.ActiveClusterID, nil
|
|
}
|
|
|
|
func normalizeTelegram(cfg Telegram) Telegram {
|
|
cfg.Token = strings.TrimSpace(cfg.Token)
|
|
cfg.AllowedUserIDs = normalizeUserIDs(cfg.AllowedUserIDs)
|
|
if cfg.Token == "" || len(cfg.AllowedUserIDs) == 0 {
|
|
cfg.Enabled = false
|
|
}
|
|
return cfg
|
|
}
|
|
|
|
func validateTelegramInput(cfg Telegram) (Telegram, error) {
|
|
cfg = normalizeTelegram(cfg)
|
|
if cfg.Token == "" {
|
|
return Telegram{}, errors.New("telegram token is required")
|
|
}
|
|
if len(cfg.AllowedUserIDs) == 0 {
|
|
return Telegram{}, errors.New("at least one telegram user id is required")
|
|
}
|
|
if cfg.Enabled && cfg.Token == "" {
|
|
return Telegram{}, errors.New("telegram cannot be enabled without token")
|
|
}
|
|
return cfg, nil
|
|
}
|
|
|
|
func normalizeUserIDs(ids []int64) []int64 {
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
seen := make(map[int64]struct{}, len(ids))
|
|
out := make([]int64, 0, len(ids))
|
|
for _, id := range ids {
|
|
if id <= 0 {
|
|
continue
|
|
}
|
|
if _, ok := seen[id]; ok {
|
|
continue
|
|
}
|
|
seen[id] = struct{}{}
|
|
out = append(out, id)
|
|
}
|
|
if len(out) == 0 {
|
|
return nil
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
|
return out
|
|
}
|
|
|
|
func (s *Service) Current() (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
if reg.ActiveClusterID == "" {
|
|
return Cluster{}, ErrNoActiveCluster
|
|
}
|
|
|
|
cluster, _, err := findCluster(reg, reg.ActiveClusterID)
|
|
if err != nil {
|
|
return Cluster{}, ErrNoActiveCluster
|
|
}
|
|
return cluster, nil
|
|
}
|
|
|
|
func (s *Service) Get(selector string) (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
cluster, _, err := findCluster(reg, selector)
|
|
return cluster, err
|
|
}
|
|
|
|
func (s *Service) Use(selector string) (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
cluster, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
reg.ActiveClusterID = reg.Clusters[idx].ID
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
return cluster, nil
|
|
}
|
|
|
|
func (s *Service) Disconnect(selector string) (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
cluster, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
reg.Clusters = append(reg.Clusters[:idx], reg.Clusters[idx+1:]...)
|
|
if reg.ActiveClusterID == cluster.ID {
|
|
reg.ActiveClusterID = ""
|
|
if len(reg.Clusters) > 0 {
|
|
reg.ActiveClusterID = reg.Clusters[0].ID
|
|
}
|
|
}
|
|
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
s.CloseTunnelClient(cluster.ID)
|
|
return cluster, nil
|
|
}
|
|
|
|
func (s *Service) ProbeSSH(ctx context.Context, selector, password, keyPassphrase string) (SSHProbeResult, error) {
|
|
cluster, err := s.Get(selector)
|
|
if err != nil {
|
|
return SSHProbeResult{}, err
|
|
}
|
|
return s.probeSSH(ctx, cluster, password, keyPassphrase), nil
|
|
}
|
|
|
|
// UpdateAuthOptions is a partial update of a cluster's authentication settings.
|
|
// Nil pointer fields mean "leave as-is"; non-nil means "replace with this value".
|
|
type UpdateAuthOptions struct {
|
|
AuthMethod AuthMethod
|
|
SetPassword bool
|
|
Password string
|
|
StorePassword bool
|
|
ClearPassword bool
|
|
SetKeyPath bool
|
|
KeyPath string
|
|
SetKeyPassphrase bool
|
|
KeyPassphrase string
|
|
StoreKeyPassphrase bool
|
|
SetKeyPassphraseFile bool
|
|
KeyPassphraseFile string
|
|
StoreKeyPassphraseFile bool
|
|
ClearKeyPassphrase bool
|
|
SetInsecureHostKey bool
|
|
InsecureHostKey bool
|
|
SetTransport bool
|
|
Transport TransportMode
|
|
}
|
|
|
|
// UpdateAuth mutates stored credentials for a cluster and persists.
|
|
func (s *Service) UpdateAuth(selector string, opts UpdateAuthOptions) (Cluster, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
c, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
|
|
if opts.AuthMethod != "" {
|
|
if opts.AuthMethod != AuthMethodPassword && opts.AuthMethod != AuthMethodKey {
|
|
return Cluster{}, fmt.Errorf("unsupported auth method %q", opts.AuthMethod)
|
|
}
|
|
c.AuthMethod = opts.AuthMethod
|
|
}
|
|
|
|
if opts.ClearPassword {
|
|
c.Password = ""
|
|
} else if opts.SetPassword {
|
|
if opts.StorePassword {
|
|
c.Password = opts.Password
|
|
} else {
|
|
c.Password = ""
|
|
}
|
|
}
|
|
|
|
if opts.SetKeyPath {
|
|
if strings.TrimSpace(opts.KeyPath) == "" {
|
|
c.KeyPath = ""
|
|
} else {
|
|
expanded, err := expandPath(opts.KeyPath)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
if _, err := os.Stat(expanded); err != nil {
|
|
return Cluster{}, fmt.Errorf("read key file: %w", err)
|
|
}
|
|
c.KeyPath = expanded
|
|
}
|
|
}
|
|
|
|
if opts.ClearKeyPassphrase {
|
|
c.KeyPassphrase = ""
|
|
c.KeyPassphraseFile = ""
|
|
} else if opts.SetKeyPassphrase {
|
|
if opts.StoreKeyPassphrase {
|
|
c.KeyPassphrase = opts.KeyPassphrase
|
|
} else {
|
|
c.KeyPassphrase = ""
|
|
}
|
|
}
|
|
if opts.SetKeyPassphraseFile {
|
|
if opts.StoreKeyPassphraseFile {
|
|
expanded, err := expandPath(opts.KeyPassphraseFile)
|
|
if err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
if _, err := os.Stat(expanded); err != nil {
|
|
return Cluster{}, fmt.Errorf("read key passphrase file: %w", err)
|
|
}
|
|
c.KeyPassphraseFile = expanded
|
|
} else {
|
|
c.KeyPassphraseFile = ""
|
|
}
|
|
}
|
|
|
|
if opts.SetInsecureHostKey {
|
|
c.InsecureHostKey = opts.InsecureHostKey
|
|
}
|
|
|
|
if opts.SetTransport {
|
|
t := normalizeTransport(opts.Transport)
|
|
if t != TransportDirect && t != TransportIPFabric {
|
|
return Cluster{}, fmt.Errorf("unsupported transport %q", opts.Transport)
|
|
}
|
|
c.Transport = t
|
|
}
|
|
|
|
// Sanity: password auth needs a password when used live; allow empty if
|
|
// the caller is only flipping method and will provide password later.
|
|
switch c.AuthMethod {
|
|
case AuthMethodKey:
|
|
if strings.TrimSpace(c.KeyPath) == "" {
|
|
return Cluster{}, errors.New("key auth requires key_path; set it with --key-path")
|
|
}
|
|
case AuthMethodPassword:
|
|
// Password may be stored or supplied at runtime; no hard check here.
|
|
default:
|
|
return Cluster{}, fmt.Errorf("unsupported auth method %q", c.AuthMethod)
|
|
}
|
|
|
|
c.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = c
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, err
|
|
}
|
|
s.CloseTunnelClient(c.ID)
|
|
return c, nil
|
|
}
|
|
|
|
// InteractiveShellOptions controls an interactive SSH session.
|
|
type InteractiveShellOptions struct {
|
|
Stdin io.Reader
|
|
Stdout io.Writer
|
|
Stderr io.Writer
|
|
Term string
|
|
Width int
|
|
Height int
|
|
Resize <-chan InteractiveShellSize
|
|
Command string // optional remote command; empty means interactive shell
|
|
}
|
|
|
|
// InteractiveShellSize carries terminal resize events.
|
|
type InteractiveShellSize struct {
|
|
Width int
|
|
Height int
|
|
}
|
|
|
|
// OpenInteractiveShell opens an interactive SSH session against the selected
|
|
// cluster using stored credentials. The caller is responsible for putting its
|
|
// local stdin into raw mode and delivering SIGWINCH events through opts.Resize.
|
|
func (s *Service) OpenInteractiveShell(ctx context.Context, selector string, opts InteractiveShellOptions) error {
|
|
c, err := s.Get(selector)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
client, err := s.dialSSH(ctx, c, "", "")
|
|
if err != nil {
|
|
return fmt.Errorf("ssh dial: %w", err)
|
|
}
|
|
defer client.Close()
|
|
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
return fmt.Errorf("ssh session: %w", err)
|
|
}
|
|
defer session.Close()
|
|
|
|
termName := strings.TrimSpace(opts.Term)
|
|
if termName == "" {
|
|
termName = os.Getenv("TERM")
|
|
}
|
|
if termName == "" {
|
|
termName = "xterm-256color"
|
|
}
|
|
|
|
width := opts.Width
|
|
height := opts.Height
|
|
if width <= 0 {
|
|
width = 120
|
|
}
|
|
if height <= 0 {
|
|
height = 32
|
|
}
|
|
|
|
modes := ssh.TerminalModes{
|
|
ssh.ECHO: 1,
|
|
ssh.ICANON: 1,
|
|
ssh.ISIG: 1,
|
|
ssh.ICRNL: 1,
|
|
ssh.OPOST: 1,
|
|
ssh.TTY_OP_ISPEED: 38400,
|
|
ssh.TTY_OP_OSPEED: 38400,
|
|
}
|
|
|
|
if err := session.RequestPty(termName, height, width, modes); err != nil {
|
|
return fmt.Errorf("ssh pty: %w", err)
|
|
}
|
|
|
|
session.Stdout = opts.Stdout
|
|
session.Stderr = opts.Stderr
|
|
stdinPipe, err := session.StdinPipe()
|
|
if err != nil {
|
|
return fmt.Errorf("ssh stdin: %w", err)
|
|
}
|
|
|
|
if strings.TrimSpace(opts.Command) == "" {
|
|
if err := session.Shell(); err != nil {
|
|
return fmt.Errorf("ssh shell: %w", err)
|
|
}
|
|
} else {
|
|
if err := session.Start(opts.Command); err != nil {
|
|
return fmt.Errorf("ssh start: %w", err)
|
|
}
|
|
}
|
|
|
|
doneCtx, cancelDone := context.WithCancel(context.Background())
|
|
defer cancelDone()
|
|
|
|
if opts.Resize != nil {
|
|
go func() {
|
|
for {
|
|
select {
|
|
case <-doneCtx.Done():
|
|
return
|
|
case sz, ok := <-opts.Resize:
|
|
if !ok {
|
|
return
|
|
}
|
|
if sz.Width <= 0 || sz.Height <= 0 {
|
|
continue
|
|
}
|
|
_ = session.WindowChange(sz.Height, sz.Width)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
copyDone := make(chan struct{})
|
|
if opts.Stdin != nil {
|
|
go func() {
|
|
_, _ = io.Copy(stdinPipe, opts.Stdin)
|
|
_ = stdinPipe.Close()
|
|
close(copyDone)
|
|
}()
|
|
} else {
|
|
close(copyDone)
|
|
}
|
|
|
|
waitErr := make(chan error, 1)
|
|
go func() {
|
|
waitErr <- session.Wait()
|
|
}()
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
_ = session.Signal(ssh.SIGHUP)
|
|
_ = session.Close()
|
|
<-waitErr
|
|
return ctx.Err()
|
|
case err := <-waitErr:
|
|
if err != nil {
|
|
var exitErr *ssh.ExitError
|
|
if errors.As(err, &exitErr) {
|
|
return nil
|
|
}
|
|
if errors.Is(err, io.EOF) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// SoftwareScan refreshes software/plugin detection for a cluster and persists it.
|
|
func (s *Service) SoftwareScan(ctx context.Context, selector, password, keyPassphrase string) (Cluster, SoftwareInfo, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, SoftwareInfo{}, err
|
|
}
|
|
|
|
cluster, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, SoftwareInfo{}, err
|
|
}
|
|
|
|
info, err := s.probeSoftware(ctx, cluster, password, keyPassphrase)
|
|
if err != nil {
|
|
return Cluster{}, SoftwareInfo{}, err
|
|
}
|
|
|
|
cluster.Software = info
|
|
cluster.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = cluster
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, SoftwareInfo{}, err
|
|
}
|
|
|
|
return cluster, info, nil
|
|
}
|
|
|
|
// InteractiveSession represents an embedded SSH PTY session owned by the caller.
|
|
type InteractiveSession struct {
|
|
client *ssh.Client
|
|
session *ssh.Session
|
|
stdin io.WriteCloser
|
|
stdout io.Reader
|
|
doneCh chan error
|
|
closeOnce sync.Once
|
|
closeErr error
|
|
}
|
|
|
|
// Read reads a chunk of PTY output from the remote session.
|
|
func (s *InteractiveSession) Read(p []byte) (int, error) {
|
|
if s == nil || s.stdout == nil {
|
|
return 0, io.EOF
|
|
}
|
|
return s.stdout.Read(p)
|
|
}
|
|
|
|
// Write sends a chunk of input to the remote session's stdin.
|
|
func (s *InteractiveSession) Write(p []byte) (int, error) {
|
|
if s == nil || s.stdin == nil {
|
|
return 0, io.ErrClosedPipe
|
|
}
|
|
return s.stdin.Write(p)
|
|
}
|
|
|
|
// Resize notifies the remote side about a new terminal size.
|
|
func (s *InteractiveSession) Resize(cols, rows int) error {
|
|
if s == nil || s.session == nil {
|
|
return nil
|
|
}
|
|
if cols <= 0 || rows <= 0 {
|
|
return nil
|
|
}
|
|
return s.session.WindowChange(rows, cols)
|
|
}
|
|
|
|
// Wait returns a channel that is closed when the remote session exits.
|
|
func (s *InteractiveSession) Wait() <-chan error {
|
|
if s == nil {
|
|
ch := make(chan error, 1)
|
|
close(ch)
|
|
return ch
|
|
}
|
|
return s.doneCh
|
|
}
|
|
|
|
// Close tears down the session and SSH client.
|
|
func (s *InteractiveSession) Close() error {
|
|
if s == nil {
|
|
return nil
|
|
}
|
|
s.closeOnce.Do(func() {
|
|
if s.stdin != nil {
|
|
_ = s.stdin.Close()
|
|
}
|
|
if s.session != nil {
|
|
_ = s.session.Close()
|
|
}
|
|
if s.client != nil {
|
|
s.closeErr = s.client.Close()
|
|
}
|
|
})
|
|
return s.closeErr
|
|
}
|
|
|
|
// StartInteractiveShell opens an SSH connection with a PTY and returns a
|
|
// handle that the caller can drive: reading remote output, writing input,
|
|
// resizing, and closing. Unlike OpenInteractiveShell, this method does NOT
|
|
// touch os.Stdin/os.Stdout — it's meant for embedding inside the TUI.
|
|
func (s *Service) StartInteractiveShell(ctx context.Context, selector string, cols, rows int, termName string) (*InteractiveSession, error) {
|
|
c, err := s.Get(selector)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
client, err := s.dialSSH(ctx, c, "", "")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ssh dial: %w", err)
|
|
}
|
|
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh session: %w", err)
|
|
}
|
|
|
|
if cols <= 0 {
|
|
cols = 120
|
|
}
|
|
if rows <= 0 {
|
|
rows = 32
|
|
}
|
|
if strings.TrimSpace(termName) == "" {
|
|
termName = "xterm-256color"
|
|
}
|
|
|
|
modes := ssh.TerminalModes{
|
|
ssh.ECHO: 1,
|
|
ssh.ICANON: 1,
|
|
ssh.ISIG: 1,
|
|
ssh.ICRNL: 1,
|
|
ssh.OPOST: 1,
|
|
ssh.TTY_OP_ISPEED: 38400,
|
|
ssh.TTY_OP_OSPEED: 38400,
|
|
}
|
|
|
|
if err := session.RequestPty(termName, rows, cols, modes); err != nil {
|
|
_ = session.Close()
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh pty: %w", err)
|
|
}
|
|
|
|
stdinPipe, err := session.StdinPipe()
|
|
if err != nil {
|
|
_ = session.Close()
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh stdin: %w", err)
|
|
}
|
|
stdoutPipe, err := session.StdoutPipe()
|
|
if err != nil {
|
|
_ = session.Close()
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh stdout: %w", err)
|
|
}
|
|
stderrPipe, err := session.StderrPipe()
|
|
if err != nil {
|
|
_ = session.Close()
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh stderr: %w", err)
|
|
}
|
|
|
|
combined := newCombinedReader(stdoutPipe, stderrPipe)
|
|
|
|
if err := session.Shell(); err != nil {
|
|
_ = session.Close()
|
|
client.Close()
|
|
return nil, fmt.Errorf("ssh shell: %w", err)
|
|
}
|
|
|
|
doneCh := make(chan error, 1)
|
|
go func() {
|
|
doneCh <- session.Wait()
|
|
close(doneCh)
|
|
}()
|
|
|
|
return &InteractiveSession{
|
|
client: client,
|
|
session: session,
|
|
stdin: stdinPipe,
|
|
stdout: combined,
|
|
doneCh: doneCh,
|
|
}, nil
|
|
}
|
|
|
|
// combinedReader merges stderr into stdout for a PTY session.
|
|
type combinedReader struct {
|
|
ch chan combinedChunk
|
|
buf []byte
|
|
err error
|
|
}
|
|
|
|
type combinedChunk struct {
|
|
data []byte
|
|
err error
|
|
}
|
|
|
|
func newCombinedReader(streams ...io.Reader) *combinedReader {
|
|
cr := &combinedReader{ch: make(chan combinedChunk, 8)}
|
|
var wg sync.WaitGroup
|
|
for _, r := range streams {
|
|
if r == nil {
|
|
continue
|
|
}
|
|
wg.Add(1)
|
|
go func(rd io.Reader) {
|
|
defer wg.Done()
|
|
buf := make([]byte, 4096)
|
|
for {
|
|
n, err := rd.Read(buf)
|
|
if n > 0 {
|
|
cp := make([]byte, n)
|
|
copy(cp, buf[:n])
|
|
cr.ch <- combinedChunk{data: cp}
|
|
}
|
|
if err != nil {
|
|
cr.ch <- combinedChunk{err: err}
|
|
return
|
|
}
|
|
}
|
|
}(r)
|
|
}
|
|
go func() {
|
|
wg.Wait()
|
|
close(cr.ch)
|
|
}()
|
|
return cr
|
|
}
|
|
|
|
func (r *combinedReader) Read(p []byte) (int, error) {
|
|
if len(r.buf) > 0 {
|
|
n := copy(p, r.buf)
|
|
r.buf = r.buf[n:]
|
|
return n, nil
|
|
}
|
|
if r.err != nil {
|
|
return 0, r.err
|
|
}
|
|
chunk, ok := <-r.ch
|
|
if !ok {
|
|
if r.err == nil {
|
|
r.err = io.EOF
|
|
}
|
|
return 0, r.err
|
|
}
|
|
if chunk.err != nil {
|
|
r.err = chunk.err
|
|
if len(chunk.data) == 0 {
|
|
return 0, r.err
|
|
}
|
|
}
|
|
n := copy(p, chunk.data)
|
|
if n < len(chunk.data) {
|
|
r.buf = chunk.data[n:]
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// RunPluginAction executes plugin-specific command over SSH on the selected cluster.
|
|
// Supported tools: kvm, lxc, lxd, bird, frr.
|
|
func (s *Service) RunPluginAction(ctx context.Context, selector, tool, action string, args []string) (string, error) {
|
|
tool = strings.ToLower(strings.TrimSpace(tool))
|
|
c, err := s.Get(selector)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
supportKnown := softwareProbeKnown(c.Software)
|
|
if !supportKnown {
|
|
if refreshed, info, scanErr := s.SoftwareScan(ctx, c.ID, "", ""); scanErr == nil {
|
|
c = refreshed
|
|
c.Software = info
|
|
supportKnown = true
|
|
}
|
|
}
|
|
|
|
if supportKnown && !isPluginToolSupported(c.Software, tool) {
|
|
return "", fmt.Errorf("support for %s was not detected on cluster %q. If you think this is a mistake, run `cluster software scan %s` and retry", tool, c.Name, c.Name)
|
|
}
|
|
|
|
client, err := s.dialSSH(ctx, c, "", "")
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer client.Close()
|
|
|
|
script, err := pluginScript(tool, action, args)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
out, err := runRemoteCommand(ctx, client, script)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *Service) probeSSH(ctx context.Context, c Cluster, password, keyPassphrase string) SSHProbeResult {
|
|
started := time.Now()
|
|
res := SSHProbeResult{
|
|
Reachable: false,
|
|
Address: fmt.Sprintf("%s@%s:%d", c.User, c.Host, c.Port),
|
|
CheckedAt: s.now().UTC(),
|
|
}
|
|
|
|
probeCtx, cancel := context.WithTimeout(ctx, defaultProbeTO)
|
|
defer cancel()
|
|
|
|
client, err := s.dialSSH(probeCtx, c, password, keyPassphrase)
|
|
if err != nil {
|
|
res.Error = err.Error()
|
|
return res
|
|
}
|
|
defer client.Close()
|
|
|
|
res.Reachable = true
|
|
res.LatencyMS = time.Since(started).Milliseconds()
|
|
return res
|
|
}
|
|
|
|
// RunRemoteShell executes a non-interactive shell command on the selected
|
|
// cluster over SSH and returns combined stdout/stderr text.
|
|
func (s *Service) RunRemoteShell(ctx context.Context, selector, command string) (string, error) {
|
|
c, err := s.Get(selector)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
cmd := strings.TrimSpace(command)
|
|
if cmd == "" {
|
|
return "", errors.New("empty remote command")
|
|
}
|
|
client, err := s.dialSSH(ctx, c, "", "")
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer client.Close()
|
|
return runRemoteCommand(ctx, client, cmd)
|
|
}
|
|
|
|
func (s *Service) probeSoftware(ctx context.Context, c Cluster, password, keyPassphrase string) (SoftwareInfo, error) {
|
|
client, err := s.dialSSH(ctx, c, password, keyPassphrase)
|
|
if err != nil {
|
|
return SoftwareInfo{}, err
|
|
}
|
|
defer client.Close()
|
|
|
|
out, err := runRemoteCommand(ctx, client, softwareProbeScript())
|
|
if err != nil {
|
|
return SoftwareInfo{}, err
|
|
}
|
|
return parseSoftwareProbe(out, s.now().UTC()), nil
|
|
}
|
|
|
|
func (s *Service) BootstrapAgent(ctx context.Context, opts BootstrapOptions) (Cluster, BootstrapResult, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, err
|
|
}
|
|
|
|
cluster, idx, err := findCluster(reg, opts.Selector)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, err
|
|
}
|
|
|
|
listenAddress := strings.TrimSpace(opts.ListenAddress)
|
|
if listenAddress == "" {
|
|
if normalizeTransport(cluster.Transport) == TransportIPFabric {
|
|
// ipfabric nodes have no default outbound route; we reach the
|
|
// agent only through the SSH tunnel, so bind to loopback.
|
|
listenAddress = "127.0.0.1:19090"
|
|
} else {
|
|
listenAddress = defaultAgentListen
|
|
}
|
|
}
|
|
|
|
agentPort := opts.AgentPort
|
|
if agentPort == 0 {
|
|
p, err := parsePortFromListen(listenAddress)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, err
|
|
}
|
|
agentPort = p
|
|
}
|
|
|
|
sshCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
|
defer cancel()
|
|
|
|
client, err := s.dialSSH(sshCtx, cluster, opts.Password, opts.KeyPassphrase)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("ssh connect for bootstrap: %w", err)
|
|
}
|
|
defer client.Close()
|
|
|
|
unameOut, err := runRemoteCommand(sshCtx, client, "uname -s; uname -m")
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("detect remote runtime: %w", err)
|
|
}
|
|
lines := splitNonEmptyLines(unameOut)
|
|
if len(lines) < 2 {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("unexpected uname output: %q", unameOut)
|
|
}
|
|
|
|
remoteOS := strings.TrimSpace(lines[0])
|
|
remoteArch := strings.TrimSpace(lines[1])
|
|
goos, goarch, err := mapRuntime(remoteOS, remoteArch)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, err
|
|
}
|
|
|
|
localBin := strings.TrimSpace(opts.LocalAgentBin)
|
|
cleanup := func() {}
|
|
if localBin == "" {
|
|
var buildErr error
|
|
localBin, cleanup, buildErr = buildAgentBinary(goos, goarch)
|
|
if buildErr != nil {
|
|
return Cluster{}, BootstrapResult{}, buildErr
|
|
}
|
|
}
|
|
defer cleanup()
|
|
|
|
if _, err := os.Stat(localBin); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("agent binary not found: %w", err)
|
|
}
|
|
|
|
remoteHome, err := remoteHomeDir(sshCtx, client)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("resolve remote home: %w", err)
|
|
}
|
|
remoteBase := path.Join(remoteHome, ".pxmon")
|
|
remoteBin := path.Join(remoteBase, "bin", "pxmon-agent")
|
|
remoteCfg := path.Join(remoteBase, "run", "agent.json")
|
|
remoteLog := path.Join(remoteBase, "log", "agent.log")
|
|
remotePID := path.Join(remoteBase, "run", "agent.pid")
|
|
|
|
mkdirScript := fmt.Sprintf(
|
|
"mkdir -p %s %s %s",
|
|
shellQuote(path.Join(remoteBase, "bin")),
|
|
shellQuote(path.Join(remoteBase, "run")),
|
|
shellQuote(path.Join(remoteBase, "log")),
|
|
)
|
|
if _, err := runRemoteCommand(sshCtx, client, mkdirScript); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("prepare remote dirs: %w", err)
|
|
}
|
|
|
|
// Upload to a sibling ".new" path to avoid ETXTBSY when the old
|
|
// binary is still running. The start script below performs an atomic
|
|
// rename into place after stopping the previous process.
|
|
remoteBinStaging := remoteBin + ".new"
|
|
if err := uploadFile(sshCtx, client, localBin, remoteBinStaging, 0o755); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("upload agent binary: %w", err)
|
|
}
|
|
|
|
token := cluster.Agent.Token
|
|
if token == "" || opts.RotateToken {
|
|
token = randomHex(24)
|
|
}
|
|
requestSecret := strings.TrimSpace(cluster.Agent.RequestSecret)
|
|
if requestSecret == "" || opts.RotateToken {
|
|
requestSecret = randomHex(32)
|
|
}
|
|
|
|
certPEM, keyPEM, certFingerprint, err := generateAgentTLSMaterial()
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("generate agent TLS cert: %w", err)
|
|
}
|
|
remoteCert := path.Join(remoteBase, "run", "agent-cert.pem")
|
|
remoteKey := path.Join(remoteBase, "run", "agent-key.pem")
|
|
if err := uploadBytes(sshCtx, client, certPEM, remoteCert, 0o600); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("upload agent tls cert: %w", err)
|
|
}
|
|
if err := uploadBytes(sshCtx, client, keyPEM, remoteKey, 0o600); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("upload agent tls key: %w", err)
|
|
}
|
|
|
|
cfgPayload, err := json.MarshalIndent(map[string]any{
|
|
"listen_addr": listenAddress,
|
|
"token": token,
|
|
"request_secret": requestSecret,
|
|
"tls_enabled": true,
|
|
"tls_cert_path": remoteCert,
|
|
"tls_key_path": remoteKey,
|
|
}, "", " ")
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("encode remote agent config: %w", err)
|
|
}
|
|
cfgPayload = append(cfgPayload, '\n')
|
|
|
|
if err := uploadBytes(sshCtx, client, cfgPayload, remoteCfg, 0o600); err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("upload agent config: %w", err)
|
|
}
|
|
|
|
// Stop the previous agent (if any), give it a brief moment to release
|
|
// the text segment, then atomically rename the staged binary into place
|
|
// and start the new process. `mv` on the same filesystem is atomic and
|
|
// safe even if the old inode is still held by the running process —
|
|
// the path is repointed to the new inode and the old one is unlinked
|
|
// once the process exits.
|
|
startScript := fmt.Sprintf(`if [ -f %s ]; then
|
|
OLDPID="$(cat %s)"
|
|
if [ -n "$OLDPID" ]; then
|
|
kill "$OLDPID" >/dev/null 2>&1 || true
|
|
for i in 1 2 3 4 5 6 7 8 9 10; do
|
|
kill -0 "$OLDPID" >/dev/null 2>&1 || break
|
|
sleep 0.2
|
|
done
|
|
kill -9 "$OLDPID" >/dev/null 2>&1 || true
|
|
fi
|
|
fi
|
|
# If any process is still listening on the target agent port (for example
|
|
# from a previous/legacy install path), stop it so the new agent can bind.
|
|
if command -v ss >/dev/null 2>&1; then
|
|
for P in $(ss -lntp 2>/dev/null | grep -E "[:.]%d[[:space:]]" | sed -n 's/.*pid=\([0-9][0-9]*\).*/\1/p' | sort -u); do
|
|
[ -n "$P" ] || continue
|
|
kill "$P" >/dev/null 2>&1 || true
|
|
for i in 1 2 3 4 5; do
|
|
kill -0 "$P" >/dev/null 2>&1 || break
|
|
sleep 0.2
|
|
done
|
|
kill -9 "$P" >/dev/null 2>&1 || true
|
|
done
|
|
fi
|
|
mv -f %s %s
|
|
chmod 0755 %s
|
|
nohup %s --config %s > %s 2>&1 &
|
|
echo $! > %s
|
|
cat %s`,
|
|
shellQuote(remotePID), shellQuote(remotePID),
|
|
agentPort,
|
|
shellQuote(remoteBinStaging), shellQuote(remoteBin),
|
|
shellQuote(remoteBin),
|
|
shellQuote(remoteBin), shellQuote(remoteCfg), shellQuote(remoteLog),
|
|
shellQuote(remotePID), shellQuote(remotePID))
|
|
pidOut, err := runRemoteCommand(sshCtx, client, startScript)
|
|
if err != nil {
|
|
return Cluster{}, BootstrapResult{}, fmt.Errorf("start remote agent: %w", err)
|
|
}
|
|
pid := strings.TrimSpace(pidOut)
|
|
|
|
cluster.Agent = AgentInstall{
|
|
Installed: true,
|
|
Version: expectedAgentVersion(),
|
|
RemoteBinary: remoteBin,
|
|
RemoteConfig: remoteCfg,
|
|
RemoteLog: remoteLog,
|
|
RemotePIDFile: remotePID,
|
|
ListenAddress: listenAddress,
|
|
Port: agentPort,
|
|
Token: token,
|
|
RequestSecret: requestSecret,
|
|
TLSEnabled: true,
|
|
TLSCertPath: remoteCert,
|
|
TLSKeyPath: remoteKey,
|
|
TLSFingerprint: certFingerprint,
|
|
LastBootstrapAt: s.now().UTC(),
|
|
}
|
|
cluster.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = cluster
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, BootstrapResult{}, err
|
|
}
|
|
|
|
s.CloseTunnelClient(cluster.ID)
|
|
ping := s.probeAgent(ctx, cluster)
|
|
if strings.TrimSpace(ping.Version) != "" && ping.Version != cluster.Agent.Version {
|
|
cluster.Agent.Version = ping.Version
|
|
cluster.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = cluster
|
|
_ = s.store.Save(reg)
|
|
}
|
|
if !ping.Reachable && !opts.AllowAgentProbe {
|
|
return Cluster{}, BootstrapResult{RemoteOS: remoteOS, RemoteArch: remoteArch, PID: pid, AgentPing: ping}, fmt.Errorf("agent deployed but probe failed: %s", ping.Error)
|
|
}
|
|
|
|
result := BootstrapResult{
|
|
RemoteOS: remoteOS,
|
|
RemoteArch: remoteArch,
|
|
PID: pid,
|
|
AgentPing: ping,
|
|
}
|
|
return cluster, result, nil
|
|
}
|
|
|
|
func (s *Service) PingAgent(ctx context.Context, selector string) (AgentPingResult, error) {
|
|
cluster, err := s.Get(selector)
|
|
if err != nil {
|
|
return AgentPingResult{}, err
|
|
}
|
|
if !cluster.Agent.Installed {
|
|
return AgentPingResult{}, errors.New("agent is not installed on this cluster")
|
|
}
|
|
return s.probeAgent(ctx, cluster), nil
|
|
}
|
|
|
|
func (s *Service) AdoptAgentAuth(ctx context.Context, selector string) (Cluster, AgentPingResult, error) {
|
|
reg, err := s.store.Load()
|
|
if err != nil {
|
|
return Cluster{}, AgentPingResult{}, err
|
|
}
|
|
c, idx, err := findCluster(reg, selector)
|
|
if err != nil {
|
|
return Cluster{}, AgentPingResult{}, err
|
|
}
|
|
|
|
client, err := s.dialSSH(ctx, c, "", "")
|
|
if err != nil {
|
|
return Cluster{}, AgentPingResult{}, err
|
|
}
|
|
defer client.Close()
|
|
|
|
remoteCfg := strings.TrimSpace(c.Agent.RemoteConfig)
|
|
if remoteCfg == "" {
|
|
home, homeErr := remoteHomeDir(ctx, client)
|
|
if homeErr != nil {
|
|
return Cluster{}, AgentPingResult{}, fmt.Errorf("resolve remote home: %w", homeErr)
|
|
}
|
|
remoteCfg = path.Join(home, ".pxmon", "run", "agent.json")
|
|
}
|
|
cfgText, err := runRemoteCommand(ctx, client, "cat "+shellQuote(remoteCfg))
|
|
if err != nil {
|
|
return Cluster{}, AgentPingResult{}, fmt.Errorf("read remote agent config: %w", err)
|
|
}
|
|
|
|
var remote struct {
|
|
ListenAddr string `json:"listen_addr"`
|
|
Token string `json:"token"`
|
|
RequestSecret string `json:"request_secret"`
|
|
TLSEnabled bool `json:"tls_enabled"`
|
|
TLSCertPath string `json:"tls_cert_path"`
|
|
TLSKeyPath string `json:"tls_key_path"`
|
|
}
|
|
if err := json.Unmarshal([]byte(cfgText), &remote); err != nil {
|
|
return Cluster{}, AgentPingResult{}, fmt.Errorf("parse remote agent config: %w", err)
|
|
}
|
|
if strings.TrimSpace(remote.Token) == "" {
|
|
return Cluster{}, AgentPingResult{}, errors.New("remote agent config has empty token")
|
|
}
|
|
|
|
fingerprint := c.Agent.TLSFingerprint
|
|
if remote.TLSEnabled {
|
|
certPath := strings.TrimSpace(remote.TLSCertPath)
|
|
if certPath == "" {
|
|
return Cluster{}, AgentPingResult{}, errors.New("remote agent TLS enabled but cert path is empty")
|
|
}
|
|
certPEM, certErr := runRemoteCommand(ctx, client, "cat "+shellQuote(certPath))
|
|
if certErr != nil {
|
|
return Cluster{}, AgentPingResult{}, fmt.Errorf("read remote agent TLS cert: %w", certErr)
|
|
}
|
|
fp, fpErr := fingerprintCertPEM([]byte(certPEM))
|
|
if fpErr != nil {
|
|
return Cluster{}, AgentPingResult{}, fpErr
|
|
}
|
|
fingerprint = fp
|
|
}
|
|
|
|
port := c.Agent.Port
|
|
if p, pErr := parsePortFromListen(remote.ListenAddr); pErr == nil {
|
|
port = p
|
|
}
|
|
if port == 0 {
|
|
port = defaultAgentPort
|
|
}
|
|
|
|
c.Agent.Installed = true
|
|
c.Agent.Version = expectedAgentVersion()
|
|
c.Agent.RemoteConfig = remoteCfg
|
|
c.Agent.ListenAddress = remote.ListenAddr
|
|
c.Agent.Port = port
|
|
c.Agent.Token = strings.TrimSpace(remote.Token)
|
|
c.Agent.RequestSecret = strings.TrimSpace(remote.RequestSecret)
|
|
c.Agent.TLSEnabled = remote.TLSEnabled
|
|
c.Agent.TLSCertPath = strings.TrimSpace(remote.TLSCertPath)
|
|
c.Agent.TLSKeyPath = strings.TrimSpace(remote.TLSKeyPath)
|
|
c.Agent.TLSFingerprint = fingerprint
|
|
c.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = c
|
|
if err := s.store.Save(reg); err != nil {
|
|
return Cluster{}, AgentPingResult{}, err
|
|
}
|
|
|
|
s.CloseTunnelClient(c.ID)
|
|
ping := s.probeAgent(ctx, c)
|
|
if strings.TrimSpace(ping.Version) != "" {
|
|
c.Agent.Version = ping.Version
|
|
c.UpdatedAt = s.now().UTC()
|
|
reg.Clusters[idx] = c
|
|
_ = s.store.Save(reg)
|
|
}
|
|
return c, ping, nil
|
|
}
|
|
|
|
func (s *Service) AgentStats(ctx context.Context, selector string) (map[string]any, error) {
|
|
body, err := s.agentStatsBody(ctx, selector)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var payload map[string]any
|
|
if err := json.Unmarshal(body, &payload); err != nil {
|
|
return nil, err
|
|
}
|
|
return payload, nil
|
|
}
|
|
|
|
func (s *Service) AgentStatsTyped(ctx context.Context, selector string) (agent.StatsResponse, error) {
|
|
body, err := s.agentStatsBody(ctx, selector)
|
|
if err != nil {
|
|
return agent.StatsResponse{}, err
|
|
}
|
|
|
|
var payload agent.StatsResponse
|
|
if err := json.Unmarshal(body, &payload); err != nil {
|
|
return agent.StatsResponse{}, err
|
|
}
|
|
return payload, nil
|
|
}
|
|
|
|
func (s *Service) agentStatsBody(ctx context.Context, selector string) ([]byte, error) {
|
|
cluster, err := s.Get(selector)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !cluster.Agent.Installed {
|
|
return nil, errors.New("agent is not installed on this cluster")
|
|
}
|
|
|
|
ac, err := s.newAgentClient(ctx, cluster, 8*time.Second)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer ac.Close()
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, ac.target+"/api/v1/stats", nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
applyAgentRequestAuth(req, cluster)
|
|
|
|
resp, err := ac.http.Do(req)
|
|
if err != nil {
|
|
if normalizeTransport(cluster.Transport) == TransportIPFabric {
|
|
s.CloseTunnelClient(cluster.ID)
|
|
}
|
|
return nil, err
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if resp.StatusCode >= 300 {
|
|
msg := strings.TrimSpace(string(body))
|
|
if len(msg) > 200 {
|
|
msg = msg[:200]
|
|
}
|
|
return nil, fmt.Errorf("agent HTTP %d: %s", resp.StatusCode, msg)
|
|
}
|
|
return body, nil
|
|
}
|
|
|
|
func (s *Service) probeAgent(ctx context.Context, c Cluster) AgentPingResult {
|
|
result := AgentPingResult{
|
|
Reachable: false,
|
|
CheckedAt: s.now().UTC(),
|
|
}
|
|
|
|
if c.Agent.Port == 0 {
|
|
result.Endpoint = fmt.Sprintf("%s://%s:%d/api/v1/ping", agentScheme(c), c.Host, c.Agent.Port)
|
|
result.Error = "agent port is not set"
|
|
return result
|
|
}
|
|
|
|
ac, err := s.newAgentClient(ctx, c, 6*time.Second)
|
|
if err != nil {
|
|
result.Endpoint = fmt.Sprintf("%s://%s:%d/api/v1/ping", agentScheme(c), c.Host, c.Agent.Port)
|
|
result.Error = err.Error()
|
|
return result
|
|
}
|
|
defer ac.Close()
|
|
|
|
result.Endpoint = ac.target + "/api/v1/ping"
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, result.Endpoint, nil)
|
|
if err != nil {
|
|
result.Error = err.Error()
|
|
return result
|
|
}
|
|
applyAgentRequestAuth(req, c)
|
|
|
|
resp, err := ac.http.Do(req)
|
|
if err != nil {
|
|
if normalizeTransport(c.Transport) == TransportIPFabric {
|
|
s.CloseTunnelClient(c.ID)
|
|
}
|
|
result.Error = err.Error()
|
|
return result
|
|
}
|
|
defer resp.Body.Close()
|
|
body, _ := io.ReadAll(resp.Body)
|
|
|
|
result.StatusCode = resp.StatusCode
|
|
result.Reachable = true
|
|
if len(body) > 0 {
|
|
var payload struct {
|
|
Version string `json:"version"`
|
|
}
|
|
if err := json.Unmarshal(body, &payload); err == nil {
|
|
result.Version = strings.TrimSpace(payload.Version)
|
|
}
|
|
}
|
|
if resp.StatusCode >= 400 {
|
|
result.Error = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func (s *Service) ExpectedAgentVersion() string {
|
|
return expectedAgentVersion()
|
|
}
|
|
|
|
func expectedAgentVersion() string {
|
|
v := strings.TrimSpace(agent.Version)
|
|
if v == "" {
|
|
return "dev"
|
|
}
|
|
return v
|
|
}
|
|
|
|
func (s *Service) dialSSH(ctx context.Context, c Cluster, passwordOverride, keyPassphraseOverride string) (*ssh.Client, error) {
|
|
client, err := s.dialSSHOnce(ctx, c, passwordOverride, keyPassphraseOverride)
|
|
if err == nil {
|
|
return client, nil
|
|
}
|
|
|
|
// Auto-heal known_hosts problems by fetching and appending the current server key,
|
|
// then retrying once.
|
|
if c.InsecureHostKey || !isKnownHostsError(err) {
|
|
return nil, err
|
|
}
|
|
if fixErr := s.ensureKnownHostEntry(ctx, c); fixErr != nil {
|
|
return nil, fmt.Errorf("%w (auto host key update failed: %v)", err, fixErr)
|
|
}
|
|
|
|
return s.dialSSHOnce(ctx, c, passwordOverride, keyPassphraseOverride)
|
|
}
|
|
|
|
func (s *Service) dialSSHOnce(ctx context.Context, c Cluster, passwordOverride, keyPassphraseOverride string) (*ssh.Client, error) {
|
|
password := strings.TrimSpace(passwordOverride)
|
|
if password == "" {
|
|
password = c.Password
|
|
}
|
|
|
|
keyPassphrase := strings.TrimSpace(keyPassphraseOverride)
|
|
if keyPassphrase == "" {
|
|
keyPassphrase = c.KeyPassphrase
|
|
}
|
|
if keyPassphrase == "" && strings.TrimSpace(c.KeyPassphraseFile) != "" {
|
|
filePassphrase, err := readStoredKeyPassphraseFile(c.KeyPassphraseFile)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
keyPassphrase = filePassphrase
|
|
}
|
|
|
|
hostKeyCallback, err := s.hostKeyCallback(c.InsecureHostKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
auth, err := sshAuthMethods(c, password, keyPassphrase)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cfg := &ssh.ClientConfig{
|
|
User: c.User,
|
|
Auth: auth,
|
|
HostKeyCallback: hostKeyCallback,
|
|
Timeout: defaultProbeTO,
|
|
}
|
|
|
|
addr := net.JoinHostPort(c.Host, strconv.Itoa(c.Port))
|
|
dialer := &net.Dialer{Timeout: defaultProbeTO}
|
|
conn, err := dialer.DialContext(ctx, "tcp", addr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
sshConn, chans, reqs, err := ssh.NewClientConn(conn, addr, cfg)
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, err
|
|
}
|
|
|
|
return ssh.NewClient(sshConn, chans, reqs), nil
|
|
}
|
|
|
|
func (s *Service) hostKeyCallback(insecure bool) (ssh.HostKeyCallback, error) {
|
|
if insecure {
|
|
return ssh.InsecureIgnoreHostKey(), nil //nolint:gosec
|
|
}
|
|
|
|
knownHostsPath, err := knownHostsPath()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := ensureKnownHostsFile(knownHostsPath); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cb, err := knownhosts.New(knownHostsPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse known_hosts: %w", err)
|
|
}
|
|
return cb, nil
|
|
}
|
|
|
|
func knownHostsPath() (string, error) {
|
|
home, err := os.UserHomeDir()
|
|
if err != nil {
|
|
return "", fmt.Errorf("resolve home directory: %w", err)
|
|
}
|
|
return filepath.Join(home, ".ssh", "known_hosts"), nil
|
|
}
|
|
|
|
func ensureKnownHostsFile(path string) error {
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
|
|
return fmt.Errorf("create ~/.ssh directory: %w", err)
|
|
}
|
|
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
|
|
if err != nil {
|
|
return fmt.Errorf("create known_hosts: %w", err)
|
|
}
|
|
return f.Close()
|
|
}
|
|
|
|
func (s *Service) ensureKnownHostEntry(ctx context.Context, c Cluster) error {
|
|
path, err := knownHostsPath()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := ensureKnownHostsFile(path); err != nil {
|
|
return err
|
|
}
|
|
|
|
key, err := fetchHostKey(ctx, c.Host, c.Port)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
hostEntry := knownHostAddress(c.Host, c.Port)
|
|
line := knownhosts.Line([]string{hostEntry}, key)
|
|
existing, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read known_hosts: %w", err)
|
|
}
|
|
if bytes.Contains(existing, []byte(line)) {
|
|
return nil
|
|
}
|
|
|
|
f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o600)
|
|
if err != nil {
|
|
return fmt.Errorf("open known_hosts for update: %w", err)
|
|
}
|
|
defer f.Close()
|
|
|
|
if len(existing) > 0 && existing[len(existing)-1] != '\n' {
|
|
if _, err := f.Write([]byte("\n")); err != nil {
|
|
return fmt.Errorf("write known_hosts newline: %w", err)
|
|
}
|
|
}
|
|
if _, err := f.WriteString(line + "\n"); err != nil {
|
|
return fmt.Errorf("append host key to known_hosts: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func fetchHostKey(ctx context.Context, host string, port int) (ssh.PublicKey, error) {
|
|
addr := net.JoinHostPort(host, strconv.Itoa(port))
|
|
dialer := &net.Dialer{Timeout: defaultProbeTO}
|
|
conn, err := dialer.DialContext(ctx, "tcp", addr)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("dial host for key scan: %w", err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
var serverKey ssh.PublicKey
|
|
cfg := &ssh.ClientConfig{
|
|
User: "pxmon-keyscan",
|
|
Auth: []ssh.AuthMethod{ssh.Password("invalid-password")},
|
|
HostKeyCallback: func(_ string, _ net.Addr, key ssh.PublicKey) error {
|
|
serverKey = key
|
|
return nil
|
|
},
|
|
Timeout: defaultProbeTO,
|
|
}
|
|
|
|
sshConn, _, _, err := ssh.NewClientConn(conn, addr, cfg)
|
|
if sshConn != nil {
|
|
_ = sshConn.Close()
|
|
}
|
|
if serverKey == nil {
|
|
if err != nil {
|
|
return nil, fmt.Errorf("fetch host key: %w", err)
|
|
}
|
|
return nil, errors.New("fetch host key: empty server key")
|
|
}
|
|
return serverKey, nil
|
|
}
|
|
|
|
func knownHostAddress(host string, port int) string {
|
|
if port == 22 {
|
|
return host
|
|
}
|
|
return fmt.Sprintf("[%s]:%d", host, port)
|
|
}
|
|
|
|
func isKnownHostsError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
var keyErr *knownhosts.KeyError
|
|
if errors.As(err, &keyErr) {
|
|
return true
|
|
}
|
|
lower := strings.ToLower(err.Error())
|
|
return strings.Contains(lower, "knownhosts:")
|
|
}
|
|
|
|
func remoteHomeDir(ctx context.Context, client *ssh.Client) (string, error) {
|
|
out, err := runRemoteCommand(ctx, client, `printf %s "$HOME"`)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
home := strings.TrimSpace(out)
|
|
if home == "" || !strings.HasPrefix(home, "/") {
|
|
return "", fmt.Errorf("unexpected HOME value: %q", home)
|
|
}
|
|
return home, nil
|
|
}
|
|
|
|
func sshAuthMethods(c Cluster, password, keyPassphrase string) ([]ssh.AuthMethod, error) {
|
|
switch c.AuthMethod {
|
|
case AuthMethodPassword:
|
|
if password == "" {
|
|
return nil, errors.New("password auth selected but password is empty; provide --password")
|
|
}
|
|
ki := ssh.KeyboardInteractive(func(user, instruction string, questions []string, echos []bool) ([]string, error) {
|
|
answers := make([]string, len(questions))
|
|
for i := range questions {
|
|
answers[i] = password
|
|
}
|
|
return answers, nil
|
|
})
|
|
return []ssh.AuthMethod{ssh.Password(password), ki}, nil
|
|
case AuthMethodKey:
|
|
if strings.TrimSpace(c.KeyPath) == "" {
|
|
return nil, errors.New("key auth selected but key path is empty")
|
|
}
|
|
keyPath, err := expandPath(c.KeyPath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
pemBytes, err := os.ReadFile(keyPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read SSH key: %w", err)
|
|
}
|
|
|
|
var signer ssh.Signer
|
|
if keyPassphrase != "" {
|
|
signer, err = ssh.ParsePrivateKeyWithPassphrase(pemBytes, []byte(keyPassphrase))
|
|
} else {
|
|
signer, err = ssh.ParsePrivateKey(pemBytes)
|
|
}
|
|
if err != nil {
|
|
if strings.Contains(strings.ToLower(err.Error()), "encrypted") {
|
|
return nil, errors.New("encrypted SSH key requires --key-passphrase")
|
|
}
|
|
return nil, fmt.Errorf("parse SSH key: %w", err)
|
|
}
|
|
return []ssh.AuthMethod{ssh.PublicKeys(signer)}, nil
|
|
default:
|
|
return nil, fmt.Errorf("unsupported auth method %q", c.AuthMethod)
|
|
}
|
|
}
|
|
|
|
func readStoredKeyPassphraseFile(path string) (string, error) {
|
|
expanded, err := expandPath(path)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
data, err := os.ReadFile(expanded)
|
|
if err != nil {
|
|
return "", fmt.Errorf("read key passphrase file: %w", err)
|
|
}
|
|
passphrase := strings.TrimRight(string(data), "\r\n")
|
|
if passphrase == "" {
|
|
return "", errors.New("key passphrase file is empty")
|
|
}
|
|
return passphrase, nil
|
|
}
|
|
|
|
func fingerprintCertPEM(certPEM []byte) (string, error) {
|
|
block, _ := pem.Decode(certPEM)
|
|
if block == nil {
|
|
return "", errors.New("decode remote agent TLS cert: no PEM block found")
|
|
}
|
|
if block.Type != "CERTIFICATE" {
|
|
return "", fmt.Errorf("decode remote agent TLS cert: unexpected PEM type %q", block.Type)
|
|
}
|
|
sum := sha256.Sum256(block.Bytes)
|
|
return hex.EncodeToString(sum[:]), nil
|
|
}
|
|
|
|
func runRemoteCommand(ctx context.Context, client *ssh.Client, script string) (string, error) {
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer session.Close()
|
|
|
|
type output struct {
|
|
buf []byte
|
|
err error
|
|
}
|
|
|
|
ch := make(chan output, 1)
|
|
go func() {
|
|
cmd := "sh -lc " + shellQuote(script)
|
|
buf, runErr := session.CombinedOutput(cmd)
|
|
ch <- output{buf: buf, err: runErr}
|
|
}()
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return "", ctx.Err()
|
|
case out := <-ch:
|
|
if out.err != nil {
|
|
message := strings.TrimSpace(string(out.buf))
|
|
if message != "" {
|
|
return "", fmt.Errorf("%w: %s", out.err, message)
|
|
}
|
|
return "", out.err
|
|
}
|
|
return string(out.buf), nil
|
|
}
|
|
}
|
|
|
|
func uploadFile(ctx context.Context, client *ssh.Client, localPath, remotePath string, mode os.FileMode) error {
|
|
f, err := os.Open(localPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer f.Close()
|
|
|
|
return uploadReader(ctx, client, f, remotePath, mode)
|
|
}
|
|
|
|
func uploadBytes(ctx context.Context, client *ssh.Client, data []byte, remotePath string, mode os.FileMode) error {
|
|
return uploadReader(ctx, client, bytes.NewReader(data), remotePath, mode)
|
|
}
|
|
|
|
func uploadReader(ctx context.Context, client *ssh.Client, r io.Reader, remotePath string, mode os.FileMode) error {
|
|
session, err := client.NewSession()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer session.Close()
|
|
|
|
var stderr bytes.Buffer
|
|
session.Stderr = &stderr
|
|
|
|
stdin, err := session.StdinPipe()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
script := fmt.Sprintf("cat > %s && chmod %o %s", shellQuote(remotePath), mode.Perm(), shellQuote(remotePath))
|
|
if err := session.Start("sh -lc " + shellQuote(script)); err != nil {
|
|
return err
|
|
}
|
|
|
|
copyDone := make(chan error, 1)
|
|
go func() {
|
|
_, copyErr := io.Copy(stdin, r)
|
|
if closeErr := stdin.Close(); closeErr != nil && copyErr == nil {
|
|
copyErr = closeErr
|
|
}
|
|
copyDone <- copyErr
|
|
}()
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case err := <-copyDone:
|
|
if err != nil {
|
|
_ = session.Wait()
|
|
msg := strings.TrimSpace(stderr.String())
|
|
if msg != "" {
|
|
return fmt.Errorf("upload stream failed: %w: %s", err, msg)
|
|
}
|
|
return fmt.Errorf("upload stream failed: %w", err)
|
|
}
|
|
}
|
|
|
|
if err := session.Wait(); err != nil {
|
|
msg := strings.TrimSpace(stderr.String())
|
|
if msg != "" {
|
|
return fmt.Errorf("%w: %s", err, msg)
|
|
}
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func buildAgentBinary(goos, goarch string) (string, func(), error) {
|
|
if _, err := os.Stat("./cmd/pxmon-agent"); err != nil {
|
|
return "", func() {}, errors.New("cmd/pxmon-agent not found; run bootstrap from project root or provide --agent-bin")
|
|
}
|
|
|
|
tmpDir, err := os.MkdirTemp("", "pxmon-agent-build-*")
|
|
if err != nil {
|
|
return "", func() {}, err
|
|
}
|
|
|
|
bin := filepath.Join(tmpDir, "pxmon-agent")
|
|
cmd := exec.Command("go", "build", "-o", bin, "./cmd/pxmon-agent")
|
|
cmd.Env = append(os.Environ(), "CGO_ENABLED=0", "GOOS="+goos, "GOARCH="+goarch)
|
|
out, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
_ = os.RemoveAll(tmpDir)
|
|
return "", func() {}, fmt.Errorf("build agent failed: %w: %s", err, strings.TrimSpace(string(out)))
|
|
}
|
|
|
|
cleanup := func() { _ = os.RemoveAll(tmpDir) }
|
|
return bin, cleanup, nil
|
|
}
|
|
|
|
func mapRuntime(unameOS, unameArch string) (string, string, error) {
|
|
osName := strings.ToLower(strings.TrimSpace(unameOS))
|
|
archName := strings.ToLower(strings.TrimSpace(unameArch))
|
|
|
|
var goos string
|
|
switch osName {
|
|
case "linux":
|
|
goos = "linux"
|
|
case "darwin":
|
|
goos = "darwin"
|
|
default:
|
|
return "", "", fmt.Errorf("unsupported remote OS %q", unameOS)
|
|
}
|
|
|
|
var goarch string
|
|
switch archName {
|
|
case "x86_64", "amd64":
|
|
goarch = "amd64"
|
|
case "i386", "i686", "386":
|
|
goarch = "386"
|
|
case "aarch64", "arm64":
|
|
goarch = "arm64"
|
|
case "armv7l", "armv6l", "arm":
|
|
goarch = "arm"
|
|
default:
|
|
return "", "", fmt.Errorf("unsupported remote architecture %q", unameArch)
|
|
}
|
|
|
|
return goos, goarch, nil
|
|
}
|
|
|
|
func parsePortFromListen(listenAddr string) (int, error) {
|
|
addr := strings.TrimSpace(listenAddr)
|
|
if addr == "" {
|
|
return defaultAgentPort, nil
|
|
}
|
|
|
|
host, portStr, err := net.SplitHostPort(addr)
|
|
if err != nil {
|
|
if strings.Count(addr, ":") == 1 {
|
|
parts := strings.Split(addr, ":")
|
|
host = parts[0]
|
|
portStr = parts[1]
|
|
} else {
|
|
return 0, fmt.Errorf("invalid listen address %q, expected host:port", addr)
|
|
}
|
|
}
|
|
|
|
_ = host
|
|
p, err := strconv.Atoi(portStr)
|
|
if err != nil || p < 1 || p > 65535 {
|
|
return 0, fmt.Errorf("invalid listen port in %q", addr)
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func findCluster(reg Registry, selector string) (Cluster, int, error) {
|
|
sel := strings.TrimSpace(selector)
|
|
if sel == "" {
|
|
sel = reg.ActiveClusterID
|
|
}
|
|
if sel == "" {
|
|
return Cluster{}, -1, ErrNoActiveCluster
|
|
}
|
|
|
|
for i, c := range reg.Clusters {
|
|
if c.ID == sel || strings.EqualFold(c.Name, sel) {
|
|
return c, i, nil
|
|
}
|
|
}
|
|
|
|
return Cluster{}, -1, ErrClusterNotFound
|
|
}
|
|
|
|
func existsTarget(clusters []Cluster, host string, port int, user string) bool {
|
|
for _, c := range clusters {
|
|
if strings.EqualFold(c.Host, host) && c.Port == port && strings.EqualFold(c.User, user) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func newClusterID() string {
|
|
b := make([]byte, 8)
|
|
if _, err := rand.Read(b); err != nil {
|
|
return fmt.Sprintf("clu_%d", time.Now().UnixNano())
|
|
}
|
|
return "clu_" + hex.EncodeToString(b)
|
|
}
|
|
|
|
func randomHex(byteLen int) string {
|
|
b := make([]byte, byteLen)
|
|
if _, err := rand.Read(b); err != nil {
|
|
return fmt.Sprintf("tok_%d", time.Now().UnixNano())
|
|
}
|
|
return hex.EncodeToString(b)
|
|
}
|
|
|
|
func generateAgentTLSMaterial() (certPEM []byte, keyPEM []byte, fingerprint string, err error) {
|
|
pub, priv, err := ed25519.GenerateKey(rand.Reader)
|
|
if err != nil {
|
|
return nil, nil, "", err
|
|
}
|
|
serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 62))
|
|
if err != nil {
|
|
return nil, nil, "", err
|
|
}
|
|
tpl := &x509.Certificate{
|
|
SerialNumber: serial,
|
|
Subject: pkix.Name{
|
|
CommonName: "pxmon-agent",
|
|
},
|
|
NotBefore: time.Now().UTC().Add(-10 * time.Minute),
|
|
NotAfter: time.Now().UTC().Add(3650 * 24 * time.Hour),
|
|
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
|
|
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
|
BasicConstraintsValid: true,
|
|
DNSNames: []string{"pxmon-agent", "localhost"},
|
|
IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")},
|
|
}
|
|
der, err := x509.CreateCertificate(rand.Reader, tpl, tpl, pub, priv)
|
|
if err != nil {
|
|
return nil, nil, "", err
|
|
}
|
|
keyRaw, err := x509.MarshalPKCS8PrivateKey(priv)
|
|
if err != nil {
|
|
return nil, nil, "", err
|
|
}
|
|
certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
|
|
keyPEM = pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyRaw})
|
|
sum := sha256.Sum256(der)
|
|
return certPEM, keyPEM, hex.EncodeToString(sum[:]), nil
|
|
}
|
|
|
|
func expandPath(path string) (string, error) {
|
|
p := strings.TrimSpace(path)
|
|
if p == "" {
|
|
return "", errors.New("empty path")
|
|
}
|
|
if strings.HasPrefix(p, "~") {
|
|
home, err := os.UserHomeDir()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if p == "~" {
|
|
p = home
|
|
} else {
|
|
p = filepath.Join(home, strings.TrimPrefix(p, "~/"))
|
|
}
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func splitNonEmptyLines(s string) []string {
|
|
parts := strings.Split(s, "\n")
|
|
out := make([]string, 0, len(parts))
|
|
for _, p := range parts {
|
|
p = strings.TrimSpace(p)
|
|
if p != "" {
|
|
out = append(out, p)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func softwareProbeScript() string {
|
|
return strings.Join([]string{
|
|
"probe() { if command -v \"$1\" >/dev/null 2>&1; then echo 1; else echo 0; fi; }",
|
|
"echo bird=$(probe birdc)",
|
|
"echo frr=$(probe vtysh)",
|
|
"if command -v virsh >/dev/null 2>&1 || command -v qemu-system-x86_64 >/dev/null 2>&1; then echo kvm=1; else echo kvm=0; fi",
|
|
"echo lxc=$(probe lxc)",
|
|
"echo lxd=$(probe lxd)",
|
|
"if command -v birdc >/dev/null 2>&1; then echo ver_bird=$(birdc --version 2>/dev/null | head -n1); fi",
|
|
"if command -v vtysh >/dev/null 2>&1; then echo ver_frr=$(vtysh -v 2>/dev/null | head -n1); fi",
|
|
"if command -v virsh >/dev/null 2>&1; then echo ver_kvm=$(virsh --version 2>/dev/null | head -n1); fi",
|
|
"if command -v lxc >/dev/null 2>&1; then echo ver_lxc=$(lxc --version 2>/dev/null | head -n1); fi",
|
|
"if command -v lxd >/dev/null 2>&1; then echo ver_lxd=$(lxd --version 2>/dev/null | head -n1); fi",
|
|
}, "; ")
|
|
}
|
|
|
|
func parseSoftwareProbe(out string, now time.Time) SoftwareInfo {
|
|
info := SoftwareInfo{
|
|
DetectedAt: now,
|
|
Versions: map[string]string{},
|
|
}
|
|
for _, line := range splitNonEmptyLines(out) {
|
|
parts := strings.SplitN(line, "=", 2)
|
|
if len(parts) != 2 {
|
|
continue
|
|
}
|
|
k := strings.ToLower(strings.TrimSpace(parts[0]))
|
|
v := strings.TrimSpace(parts[1])
|
|
switch k {
|
|
case "bird":
|
|
info.Bird = v == "1" || strings.EqualFold(v, "true")
|
|
case "frr":
|
|
info.FRR = v == "1" || strings.EqualFold(v, "true")
|
|
case "kvm":
|
|
info.KVM = v == "1" || strings.EqualFold(v, "true")
|
|
case "lxc":
|
|
info.LXC = v == "1" || strings.EqualFold(v, "true")
|
|
case "lxd":
|
|
info.LXD = v == "1" || strings.EqualFold(v, "true")
|
|
case "ver_bird":
|
|
if v != "" {
|
|
info.Versions["bird"] = v
|
|
}
|
|
case "ver_frr":
|
|
if v != "" {
|
|
info.Versions["frr"] = v
|
|
}
|
|
case "ver_kvm":
|
|
if v != "" {
|
|
info.Versions["kvm"] = v
|
|
}
|
|
case "ver_lxc":
|
|
if v != "" {
|
|
info.Versions["lxc"] = v
|
|
}
|
|
case "ver_lxd":
|
|
if v != "" {
|
|
info.Versions["lxd"] = v
|
|
}
|
|
}
|
|
}
|
|
if len(info.Versions) == 0 {
|
|
info.Versions = nil
|
|
}
|
|
return info
|
|
}
|
|
|
|
func softwareProbeKnown(info SoftwareInfo) bool {
|
|
if !info.DetectedAt.IsZero() {
|
|
return true
|
|
}
|
|
if info.Bird || info.FRR || info.KVM || info.LXC || info.LXD {
|
|
return true
|
|
}
|
|
return len(info.Versions) > 0
|
|
}
|
|
|
|
func isPluginToolSupported(info SoftwareInfo, tool string) bool {
|
|
switch strings.ToLower(strings.TrimSpace(tool)) {
|
|
case "bird":
|
|
return info.Bird
|
|
case "frr":
|
|
return info.FRR
|
|
case "kvm":
|
|
return info.KVM
|
|
case "lxc":
|
|
return info.LXC || info.LXD
|
|
case "lxd":
|
|
return info.LXD || info.LXC
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func pluginScript(tool, action string, args []string) (string, error) {
|
|
tool = strings.ToLower(strings.TrimSpace(tool))
|
|
action = strings.ToLower(strings.TrimSpace(action))
|
|
|
|
switch tool {
|
|
case "kvm":
|
|
return kvmPluginScript(action, args)
|
|
case "lxc", "lxd":
|
|
return lxcPluginScript(action, args)
|
|
case "bird":
|
|
return birdPluginScript(action)
|
|
case "frr":
|
|
return frrPluginScript(action)
|
|
default:
|
|
return "", fmt.Errorf("unsupported plugin tool %q", tool)
|
|
}
|
|
}
|
|
|
|
func kvmPluginScript(action string, args []string) (string, error) {
|
|
mustDomain := func() (string, error) {
|
|
if len(args) < 1 || strings.TrimSpace(args[0]) == "" {
|
|
return "", errors.New("domain is required")
|
|
}
|
|
return shellQuote(strings.TrimSpace(args[0])), nil
|
|
}
|
|
|
|
switch action {
|
|
case "", "list", "domains":
|
|
return "virsh list --all", nil
|
|
case "start":
|
|
d, err := mustDomain()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "virsh start " + d, nil
|
|
case "stop", "shutdown":
|
|
d, err := mustDomain()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "virsh shutdown " + d, nil
|
|
case "reboot", "restart":
|
|
d, err := mustDomain()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "virsh reboot " + d, nil
|
|
case "destroy", "force-stop":
|
|
d, err := mustDomain()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "virsh destroy " + d, nil
|
|
case "top":
|
|
return strings.Join([]string{
|
|
`names="$(mktemp)"; states="$(mktemp)"; raw="$(mktemp)"; trap 'rm -f "$names" "$states" "$raw"' EXIT`,
|
|
`virsh list --all --name 2>/dev/null | awk 'NF>0{print $1}' > "$names"`,
|
|
`virsh list --all 2>/dev/null | awk 'NR>2 && NF>=3 {name=$2; state=$3; for(i=4;i<=NF;i++) state=state" "$i; print name "\t" state}' > "$states"`,
|
|
`while IFS= read -r d; do [ -z "$d" ] && continue; state=$(awk -F'\t' -v n="$d" '$1==n{print $2; found=1; exit} END{if(!found) print "-"}' "$states"); info=$(virsh dominfo "$d" 2>/dev/null || true); vcpu=$(printf "%s\n" "$info" | awk -F: '/^CPU\(s\):/{gsub(/[[:space:]]+/,"",$2); print $2; exit}'); ram_kib=$(printf "%s\n" "$info" | awk -F: '/^Max memory:/{gsub(/[^0-9]/,"",$2); print $2; exit}'); [ -z "$vcpu" ] && vcpu=-1; [ -z "$ram_kib" ] && ram_kib=-1; disk_cap=0; disk_alloc=0; has_disk=0; dl="$(mktemp)"; virsh domblklist "$d" --details 2>/dev/null | awk 'NR>2 && $2=="disk" && $3!=""{t=$3; s=$4; if(s=="") s="-"; print t "\t" s}' > "$dl"; tab="$(printf '\t')"; while IFS="$tab" read -r dev src; do [ -z "$dev" ] && continue; binfo=$(virsh domblkinfo "$d" "$dev" 2>/dev/null || true); cap=$(printf "%s\n" "$binfo" | awk '/^Capacity:/{print $2; exit}'); alloc=$(printf "%s\n" "$binfo" | awk '/^Allocation:/{print $2; exit}'); if [ -z "$cap" ] && [ -n "$src" ] && [ "$src" != "-" ]; then if command -v qemu-img >/dev/null 2>&1; then cap=$(qemu-img info --output=json "$src" 2>/dev/null | awk -F: '/"virtual-size"[[:space:]]*:/{gsub(/[^0-9]/,"",$2); print $2; exit}'); if [ -z "$cap" ]; then cap=$(qemu-img info "$src" 2>/dev/null | awk -F'[()]' '/virtual size:/{gsub(/[^0-9]/,"",$2); print $2; exit}'); fi; fi; fi; if [ -z "$alloc" ] && [ -n "$src" ] && [ "$src" != "-" ] && [ -f "$src" ]; then alloc=$(wc -c < "$src" 2>/dev/null | tr -d '[:space:]'); fi; if [ -n "$cap" ] || [ -n "$alloc" ]; then has_disk=1; [ -z "$cap" ] && cap=0; [ -z "$alloc" ] && alloc=0; disk_cap=$((disk_cap + cap)); disk_alloc=$((disk_alloc + alloc)); fi; done < "$dl"; rm -f "$dl"; if [ "$has_disk" -eq 0 ]; then disk_cap=-1; disk_alloc=-1; fi; printf "%s\t%s\t%s\t%s\t%s\t%s\n" "$d" "$state" "$vcpu" "$ram_kib" "$disk_cap" "$disk_alloc" >> "$raw"; done < "$names"`,
|
|
`sort -t "$(printf '\t')" -k1,1 "$raw" | awk -F'\t' 'BEGIN{printf "%-34s %-12s %7s %11s %11s %11s\n","DOMAIN","STATE","VCPU","RAM_MAX","DISK_CAP","DISK_ALLOC"} {dom=fit($1,34); st=fit($2,12); vc=numOrDash($3); ram=humanKiBOrDash($4); dcap=humanBytesOrDash($5); dalloc=humanBytesOrDash($6); printf "%-34s %-12s %7s %11s %11s %11s\n",dom,st,vc,ram,dcap,dalloc} function numOrDash(v){if(v==""||v<0)return "-"; return sprintf("%d", v+0)} function humanKiBOrDash(v){if(v==""||v<0)return "-"; return human((v+0)*1024)} function humanBytesOrDash(v){if(v==""||v<0)return "-"; return human(v+0)} function human(n, u){if(n==0)return "0B"; split("B KiB MiB GiB TiB PiB",u," "); i=1; while(n>=1024 && i<6){n/=1024; i++} if(n>=10 || i==1) return sprintf("%.0f%s",n,u[i]); return sprintf("%.1f%s",n,u[i]);} function fit(s,w){if(length(s)<=w)return s; return substr(s,1,w-1)"…"}'`,
|
|
}, "; "), nil
|
|
case "net-top", "net":
|
|
return strings.Join([]string{
|
|
`names="$(mktemp)"; states="$(mktemp)"; prev="$(mktemp)"; curr="$(mktemp)"; samples="$(mktemp)"; merged="$(mktemp)"; trap 'rm -f "$names" "$states" "$prev" "$curr" "$samples" "$merged"' EXIT`,
|
|
`virsh list --all --name 2>/dev/null | awk 'NF>0{print $1}' > "$names"`,
|
|
`virsh list --all 2>/dev/null | awk 'NR>2 && NF>=3 {name=$2; state=$3; for(i=4;i<=NF;i++) state=state" "$i; print name "\t" state}' > "$states"`,
|
|
`collect_net_snapshot() { out="$1"; virsh domstats --raw --interface 2>/dev/null | awk 'BEGIN{OFS="\t"; dom=""; q=sprintf("%c",39)} /^Domain:[[:space:]]+/ {dom=$2; if(substr(dom,1,1)==q) dom=substr(dom,2); if(length(dom)>0 && substr(dom,length(dom),1)==q) dom=substr(dom,1,length(dom)-1); seen[dom]=1; next} dom==""{next} /^net\.[0-9]+\.rx\.bytes=/{split($0,a,"="); rx[dom]+=a[2]; hasRx[dom]=1; next} /^net\.[0-9]+\.tx\.bytes=/{split($0,a,"="); tx[dom]+=a[2]; hasTx[dom]=1; next} END{for(d in seen){nr=(hasRx[d]!=""?rx[d]:-1); nt=(hasTx[d]!=""?tx[d]:-1); print d, nr, nt}}' > "$out"; }`,
|
|
`collect_net_snapshot "$prev"; : > "$samples"; loops=4; intv=0.3; i=1; while [ "$i" -le "$loops" ]; do sleep "$intv"; collect_net_snapshot "$curr"; awk -F'\t' -v intv="$intv" 'NR==FNR{prx[$1]=$2; ptx[$1]=$3; next} {d=$1; rx=$2+0; tx=$3+0; if(!(d in prx)||prx[d]<0||ptx[d]<0||rx<0||tx<0){next} drx=rx-prx[d]; if(drx<0)drx=0; dtx=tx-ptx[d]; if(dtx<0)dtx=0; rxm=drx*8/intv/1000000; txm=dtx*8/intv/1000000; tot=rxm+txm; printf "%s\t%.6f\t%.6f\t%.6f\t%.0f\t%.0f\n", d, tot, rxm, txm, rx, tx}' "$prev" "$curr" >> "$samples"; cp "$curr" "$prev"; i=$((i+1)); done`,
|
|
`awk -F'\t' 'NR==FNR{st[$1]=$2; next} {d=$1; n[d]++; tot[d,n[d]]=$2+0; rx[d]=$3+0; tx[d]=$4+0; rxTot[d]=$5+0; txTot[d]=$6+0; seen[d]=1} END{while((getline dom < "'"$names"'")>0){if(dom=="")continue; s=st[dom]; if(s=="") s="-"; if(seen[dom]==""){print dom"\t"s"\t-1\t-1\t-1\t-1\t-1\t-1"; continue} k=n[dom]; delete arr; for(i=1;i<=k;i++) arr[i]=tot[dom,i]; for(i=1;i<=k;i++) for(j=i+1;j<=k;j++) if(arr[i]>arr[j]){tmp=arr[i]; arr[i]=arr[j]; arr[j]=tmp} idx=int(0.95*k); if((0.95*k)>idx) idx++; if(idx<1) idx=1; if(idx>k) idx=k; p95=arr[idx]; print dom"\t"s"\t"tot[dom,k]"\t"rx[dom]"\t"tx[dom]"\t"p95"\t"rxTot[dom]"\t"txTot[dom]}}' "$states" "$samples" > "$merged"`,
|
|
`sort -t "$(printf '\t')" -k3,3nr -k1,1 "$merged" | awk -F'\t' 'BEGIN{printf "%-34s %-12s %10s %10s %10s %10s %11s %11s\n","DOMAIN","STATE","NET_Mbps","RX_Mbps","TX_Mbps","P95_Mbps","RX_TOTAL","TX_TOTAL"} {dom=fit($1,34); st=fit($2,12); net=mbpsOrDash($3); rxm=mbpsOrDash($4); txm=mbpsOrDash($5); p95=mbpsOrDash($6); r=humanOrDash($7); t=humanOrDash($8); printf "%-34s %-12s %10s %10s %10s %10s %11s %11s\n",dom,st,net,rxm,txm,p95,r,t} function mbpsOrDash(v){if(v==""||v<0)return "-"; return sprintf("%.2f",v+0)} function humanOrDash(v){if(v==""||v<0)return "-"; return human(v+0)} function human(n, u){if(n==0)return "0B"; if(n<0)return "-"; split("B KiB MiB GiB TiB PiB",u," "); i=1; while(n>=1024 && i<6){n/=1024; i++} if(n>=10 || i==1) return sprintf("%.0f%s",n,u[i]); return sprintf("%.1f%s",n,u[i]);} function fit(s,w){if(length(s)<=w)return s; return substr(s,1,w-1)"…"}'`,
|
|
}, "; "), nil
|
|
default:
|
|
return "", fmt.Errorf("unsupported kvm action %q", action)
|
|
}
|
|
}
|
|
|
|
func lxcPluginScript(action string, args []string) (string, error) {
|
|
mustName := func() (string, error) {
|
|
if len(args) < 1 || strings.TrimSpace(args[0]) == "" {
|
|
return "", errors.New("instance name is required")
|
|
}
|
|
return shellQuote(strings.TrimSpace(args[0])), nil
|
|
}
|
|
|
|
switch action {
|
|
case "", "list", "ls", "status":
|
|
return strings.Join([]string{
|
|
`printf "INSTANCE\tSTATE\tTYPE\tIPV4\tSNAPSHOTS\n"`,
|
|
`lxc list --format csv -c ns4tS 2>/dev/null | awk -F, 'NF>=1 {name=$1;state=$2;ipv4=$3;typ=$4;snap=$5; if(name=="") next; if(state=="") state="-"; if(typ=="") typ="-"; if(ipv4=="") ipv4="-"; if(snap=="") snap="0"; printf "%s\t%s\t%s\t%s\t%s\n", name, state, typ, ipv4, snap}'`,
|
|
}, "; "), nil
|
|
case "start":
|
|
n, err := mustName()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "lxc start " + n, nil
|
|
case "stop":
|
|
n, err := mustName()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "lxc stop " + n, nil
|
|
case "restart", "reboot":
|
|
n, err := mustName()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "lxc restart " + n, nil
|
|
case "top":
|
|
return strings.Join([]string{
|
|
`printf "INSTANCE\tSTATE\tCPU_SEC\tMEM_CUR\tMEM_PEAK\tRX\tTX\n"`,
|
|
`lxc list --format csv -c ns 2>/dev/null | while IFS=, read -r name state; do [ -z "$name" ] && continue; info=$(lxc info "$name" --resources 2>/dev/null || lxc info "$name" 2>/dev/null || true); cpu=$(printf "%s\n" "$info" | awk -F': ' '/CPU usage \(in seconds\)/{print $2; exit}'); mem_cur=$(printf "%s\n" "$info" | awk -F': ' '/Memory \(current\)/{print $2; exit}'); mem_peak=$(printf "%s\n" "$info" | awk -F': ' '/Memory \(peak\)/{print $2; exit}'); rx=$(printf "%s\n" "$info" | awk -F': ' '/Bytes received/{print $2; exit}'); tx=$(printf "%s\n" "$info" | awk -F': ' '/Bytes sent/{print $2; exit}'); [ -z "$cpu" ] && cpu=0; [ -z "$mem_cur" ] && mem_cur="-"; [ -z "$mem_peak" ] && mem_peak="-"; [ -z "$rx" ] && rx="-"; [ -z "$tx" ] && tx="-"; printf "%s\t%s\t%s\t%s\t%s\t%s\t%s\n" "$name" "${state:--}" "$cpu" "$mem_cur" "$mem_peak" "$rx" "$tx"; done | sort -k3nr`,
|
|
}, "; "), nil
|
|
case "net-top", "net":
|
|
return strings.Join([]string{
|
|
`printf "INSTANCE\tSTATE\tRX_MiB\tTX_MiB\tTOTAL_MiB\n"`,
|
|
`lxc list --format csv -c ns 2>/dev/null | while IFS=, read -r name state; do [ -z "$name" ] && continue; raw=$(lxc query "/1.0/instances/$name/state" 2>/dev/null || true); rx=$(printf "%s\n" "$raw" | awk -F: '/"bytes_received"[[:space:]]*:/{gsub(/[^0-9]/,"",$2); s+=$2} END{printf "%.0f", s+0}'); tx=$(printf "%s\n" "$raw" | awk -F: '/"bytes_sent"[[:space:]]*:/{gsub(/[^0-9]/,"",$2); s+=$2} END{printf "%.0f", s+0}'); [ -z "$rx" ] && rx=0; [ -z "$tx" ] && tx=0; total=$((rx+tx)); rx_mib=$(awk -v n="$rx" 'BEGIN{printf "%.2f", n/1048576}'); tx_mib=$(awk -v n="$tx" 'BEGIN{printf "%.2f", n/1048576}'); total_mib=$(awk -v n="$total" 'BEGIN{printf "%.2f", n/1048576}'); printf "%s\t%s\t%s\t%s\t%s\n" "$name" "${state:--}" "$rx_mib" "$tx_mib" "$total_mib"; done | sort -k5nr`,
|
|
}, "; "), nil
|
|
case "stats", "info", "show":
|
|
n, err := mustName()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "lxc info " + n + " --resources 2>/dev/null || lxc info " + n, nil
|
|
default:
|
|
return "", fmt.Errorf("unsupported lxc action %q", action)
|
|
}
|
|
}
|
|
|
|
func birdPluginScript(action string) (string, error) {
|
|
switch action {
|
|
case "", "status":
|
|
return "birdc show status", nil
|
|
case "protocols", "proto":
|
|
return "birdc show protocols", nil
|
|
case "routes", "route":
|
|
return "birdc show route", nil
|
|
default:
|
|
return "", fmt.Errorf("unsupported bird action %q", action)
|
|
}
|
|
}
|
|
|
|
func frrPluginScript(action string) (string, error) {
|
|
switch action {
|
|
case "", "status", "summary":
|
|
return "vtysh -c 'show version' -c 'show ip bgp summary'", nil
|
|
case "routes", "route":
|
|
return "vtysh -c 'show ip route summary' -c 'show ip route'", nil
|
|
case "bgp":
|
|
return "vtysh -c 'show ip bgp summary' -c 'show bgp ipv4 unicast summary'", nil
|
|
case "ospf":
|
|
return "vtysh -c 'show ip ospf neighbor' -c 'show ip ospf route'", nil
|
|
default:
|
|
return "", fmt.Errorf("unsupported frr action %q", action)
|
|
}
|
|
}
|
|
|
|
func shellQuote(v string) string {
|
|
return "'" + strings.ReplaceAll(v, "'", `'"'"'`) + "'"
|
|
}
|