Files
vantage-agent/internal/sync/sync.go
T

606 lines
16 KiB
Go

package agentsync
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"log"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"sync"
"time"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/config"
agentexec "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/exec"
grpcclient "gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/grpc"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/grpc/pb"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/inventory"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/keys"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/monitors"
"gitea.hostxtra.co.uk/mrhid6/vantage/agent/internal/updates"
)
func Run(ctx context.Context, cfg *config.Config, version string) error {
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
return fmt.Errorf("dial grpc: %w", err)
}
defer client.Close()
if cfg.PreRegToken != "" {
log.Println("registering with server...")
hostname, _ := os.Hostname()
ipAddress := localIP()
osInfo := fmt.Sprintf("%s %s", runtime.GOOS, runtime.GOARCH)
agentToken, err := client.Register(cfg.ServerID, cfg.PreRegToken, hostname, ipAddress, osInfo)
if err != nil {
return fmt.Errorf("registration failed: %w", err)
}
cfg.AgentToken = agentToken
cfg.PreRegToken = ""
if err := config.Save(cfg); err != nil {
return fmt.Errorf("save config: %w", err)
}
log.Println("registration successful")
client.Close()
client, err = grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
return fmt.Errorf("reconnect: %w", err)
}
}
if cfg.AgentToken == "" {
return fmt.Errorf("no agent token available registration required")
}
go runCommandStream(ctx, cfg)
go runUpdateCheck(ctx, cfg)
go runInventory(ctx, cfg)
go monitors.Run(ctx, cfg)
ticker := time.NewTicker(cfg.PollInterval)
defer ticker.Stop()
if err := poll(client, cfg, version); err != nil {
log.Printf("poll error: %v", err)
}
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
if err := poll(client, cfg, version); err != nil {
log.Printf("poll error: %v", err)
}
}
}
}
func poll(client *grpcclient.Client, cfg *config.Config, version string) error {
desired, err := client.SyncKeys(cfg.ServerID, cfg.AgentToken, version)
if err != nil {
return fmt.Errorf("SyncKeys: %w", err)
}
if runtime.GOOS != "linux" {
return nil
}
current, err := keys.ReadAuthorizedKeys()
if err != nil {
return fmt.Errorf("read authorized_keys: %w", err)
}
if !keys.StateChanged(current, desired) {
log.Println("authorized_keys unchanged, skipping write")
return nil
}
if err := keys.WriteAuthorizedKeys(desired); err != nil {
return fmt.Errorf("write authorized_keys: %w", err)
}
log.Printf("authorized_keys updated (%d keys)", len(desired))
return nil
}
func runCommandStream(ctx context.Context, cfg *config.Config) {
backoff := time.Second
const maxBackoff = 2 * time.Minute
for {
select {
case <-ctx.Done():
return
default:
}
if err := connectAndHandleStream(ctx, cfg); err != nil {
if ctx.Err() != nil {
return
}
log.Printf("command stream error: %v, reconnecting in %s", err, backoff)
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff < maxBackoff {
backoff *= 2
}
} else {
backoff = time.Second
}
}
}
func connectAndHandleStream(ctx context.Context, cfg *config.Config) error {
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
return fmt.Errorf("dial: %w", err)
}
defer client.Close()
stream, err := client.CommandStream(ctx)
if err != nil {
return fmt.Errorf("open stream: %w", err)
}
if err := stream.Send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
Ready: &pb.AgentReady{},
}); err != nil {
return fmt.Errorf("send auth: %w", err)
}
log.Println("command stream connected")
var sendMu sync.Mutex
send := func(msg *pb.AgentMessage) error {
sendMu.Lock()
defer sendMu.Unlock()
return stream.Send(msg)
}
for {
cmd, err := stream.Recv()
if err != nil {
return fmt.Errorf("recv: %w", err)
}
if cmd.GenerateKey != nil {
go handleGenerateKey(cfg, cmd)
}
if cmd.DeleteKey != nil {
go handleDeleteKey(cmd)
}
if cmd.UpdateAgent != nil {
go handleUpdateAgent(cmd)
}
if cmd.ApplyUpdates != nil {
go handleApplyUpdates(cfg, cmd)
}
if cmd.CleanupWorkspace != nil {
go handleCleanupWorkspace(cmd)
}
if cmd.RunStep != nil {
go func(rc *pb.RunStepCmd, cid string) {
emit := func(seq uint64, data []byte) {
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Seq: seq, Data: data},
})
}
res := agentexec.RunStep(rc, emit)
res.CommandId = cid
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepOutput: &pb.StepOutputChunk{CommandId: cid, Eof: true},
})
_ = send(&pb.AgentMessage{
ServerId: cfg.ServerID,
AgentToken: cfg.AgentToken,
StepResult: res,
})
}(cmd.RunStep, cmd.CommandId)
continue
}
}
}
func runUpdateCheck(ctx context.Context, cfg *config.Config) {
const interval = time.Hour
doCheck := func() {
pkgs, err := updates.CheckAvailable()
if err != nil {
log.Printf("update check error: %v", err)
return
}
pbUpdates := make([]pb.PackageUpdate, len(pkgs))
for i, p := range pkgs {
pbUpdates[i] = pb.PackageUpdate{
Name: p.Name,
CurrentVersion: p.CurrentVersion,
NewVersion: p.NewVersion,
}
}
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
log.Printf("update report dial error: %v", err)
return
}
defer client.Close()
if err := client.ReportUpdates(cfg.ServerID, cfg.AgentToken, pbUpdates); err != nil {
log.Printf("ReportUpdates error: %v", err)
return
}
log.Printf("reported %d available OS updates", len(pkgs))
}
doCheck()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
doCheck()
}
}
}
func runInventory(ctx context.Context, cfg *config.Config) {
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
log.Printf("inventory dial error: %v", err)
return
}
defer client.Close()
report := func(static bool) {
r := inventory.Collect(static)
r.ServerId = cfg.ServerID
r.AgentToken = cfg.AgentToken
if err := client.ReportInventory(r); err != nil {
log.Printf("report inventory: %v", err)
}
}
report(true)
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
tick := 0
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
tick++
report(tick%30 == 0)
}
}
}
func handleApplyUpdates(cfg *config.Config, cmd *pb.ServerCommand) {
log.Printf("applying OS updates (cmd=%s)…", cmd.CommandId)
if err := updates.ApplyAll(); err != nil {
log.Printf("OS upgrade failed (cmd=%s): %v", cmd.CommandId, err)
return
}
log.Printf("OS updates applied successfully (cmd=%s)", cmd.CommandId)
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
return
}
defer client.Close()
_ = client.ReportUpdates(cfg.ServerID, cfg.AgentToken, nil)
}
func handleCleanupWorkspace(cmd *pb.ServerCommand) {
id := cmd.CleanupWorkspace.WorkspaceId
dir := agentexec.WorkspacePath(id)
if err := os.RemoveAll(dir); err != nil {
log.Printf("cleanup workspace %s failed (cmd=%s): %v", dir, cmd.CommandId, err)
return
}
log.Printf("removed run workspace %s (cmd=%s)", dir, cmd.CommandId)
}
func handleDeleteKey(cmd *pb.ServerCommand) {
label := cmd.DeleteKey.Label
keyPath := fmt.Sprintf("/root/.ssh/vantage_%s", strings.ReplaceAll(label, " ", "_"))
if err := keys.RemoveSSHIdentity(keyPath); err != nil {
log.Printf("remove ssh identity failed (cmd=%s): %v", cmd.CommandId, err)
}
for _, path := range []string{keyPath, keyPath + ".pub"} {
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
log.Printf("delete key file %s (cmd=%s): %v", path, cmd.CommandId, err)
}
}
log.Printf("deleted local key files for %q (cmd=%s)", label, cmd.CommandId)
}
func handleUpdateAgent(cmd *pb.ServerCommand) {
if runtime.GOOS == "windows" {
handleUpdateAgentWindows(cmd)
return
}
u := cmd.UpdateAgent
arch := runtime.GOARCH
tag := "agent%2Fv" + u.Version
binaryURL := fmt.Sprintf("%s/mrhid6/vantage/releases/download/%s/vantage-agent-linux-%s", u.GiteaBaseURL, tag, arch)
checksumURL := fmt.Sprintf("%s/mrhid6/vantage/releases/download/%s/checksums.txt", u.GiteaBaseURL, tag)
log.Printf("updating agent to v%s from %s (cmd=%s)", u.Version, u.GiteaBaseURL, cmd.CommandId)
tmpBin := "/tmp/vantage-agent-update"
if err := downloadFile(binaryURL, tmpBin); err != nil {
log.Printf("update download failed (cmd=%s): %v", cmd.CommandId, err)
return
}
checksumData, err := httpGetBytes(checksumURL)
if err != nil {
log.Printf("update checksum fetch failed (cmd=%s): %v", cmd.CommandId, err)
return
}
if err := verifyChecksum(tmpBin, fmt.Sprintf("vantage-agent-linux-%s", arch), checksumData); err != nil {
log.Printf("update checksum mismatch (cmd=%s): %v", cmd.CommandId, err)
os.Remove(tmpBin)
return
}
if err := os.Chmod(tmpBin, 0755); err != nil {
log.Printf("update chmod failed (cmd=%s): %v", cmd.CommandId, err)
return
}
if err := os.Rename(tmpBin, "/usr/local/bin/vantage-agent"); err != nil {
log.Printf("update replace binary failed (cmd=%s): %v", cmd.CommandId, err)
return
}
log.Printf("agent binary replaced, restarting service (cmd=%s)", cmd.CommandId)
exec.Command("systemctl", "restart", "vantage-agent").Run()
}
func handleUpdateAgentWindows(cmd *pb.ServerCommand) {
u := cmd.UpdateAgent
tag := "agent%2Fv" + u.Version
msiURL := fmt.Sprintf("%s/mrhid6/vantage/releases/download/%s/vantage-agent.msi", u.GiteaBaseURL, tag)
checksumURL := fmt.Sprintf("%s/mrhid6/vantage/releases/download/%s/checksums-msi.txt", u.GiteaBaseURL, tag)
log.Printf("updating agent to v%s from %s (cmd=%s)", u.Version, u.GiteaBaseURL, cmd.CommandId)
msiPath := filepath.Join(os.TempDir(), "vantage-agent-update.msi")
if err := downloadFile(msiURL, msiPath); err != nil {
log.Printf("update download failed (cmd=%s): %v", cmd.CommandId, err)
return
}
checksumData, err := httpGetBytes(checksumURL)
if err != nil {
log.Printf("update checksum fetch failed (cmd=%s): %v", cmd.CommandId, err)
return
}
if err := verifyChecksum(msiPath, "vantage-agent.msi", checksumData); err != nil {
log.Printf("update checksum mismatch (cmd=%s): %v", cmd.CommandId, err)
os.Remove(msiPath)
return
}
logPath := filepath.Join(os.TempDir(), "vantage-agent-msi.log")
// The MSI stops the vantage-agent service as part of the upgrade. Anything
// descended from this process is killed with it, so msiexec must not be a
// child: run it from a scheduled task, which is parented to the Task
// Scheduler service instead.
if err := launchDetachedUpdate(msiPath, logPath, cmd.CommandId); err != nil {
log.Printf("failed to launch msiexec (cmd=%s): %v", cmd.CommandId, err)
return
}
log.Printf("scheduled msiexec for upgrade to v%s (cmd=%s)", u.Version, cmd.CommandId)
}
const updateTaskName = "VantageAgentUpdate"
func launchDetachedUpdate(msiPath, logPath, commandID string) error {
scriptPath := filepath.Join(os.TempDir(), "vantage-agent-update.cmd")
script := fmt.Sprintf("@echo off\r\n"+
"timeout /t 5 /nobreak >nul\r\n"+
"msiexec /i \"%s\" /qn /norestart /l*v \"%s\"\r\n"+
"schtasks /delete /tn %s /f >nul 2>&1\r\n"+
"del /f /q \"%s\" >nul 2>&1\r\n"+
"(goto) 2>nul & del /f /q \"%%~f0\"\r\n",
msiPath, logPath, updateTaskName, msiPath)
if err := os.WriteFile(scriptPath, []byte(script), 0o600); err != nil {
return fmt.Errorf("write update script: %w", err)
}
// Stale task from a previous attempt would make /create fail even with /f
// if it is still running, so tear it down first and ignore the result.
exec.Command("schtasks", "/end", "/tn", updateTaskName).Run()
exec.Command("schtasks", "/delete", "/tn", updateTaskName, "/f").Run()
create := exec.Command("schtasks", "/create",
"/tn", updateTaskName,
"/tr", `"`+scriptPath+`"`,
"/sc", "once",
// Already in the past: the task never fires on its own, only via /run.
"/st", "00:00",
"/ru", "SYSTEM",
"/rl", "HIGHEST",
"/f")
if out, err := create.CombinedOutput(); err != nil {
return fmt.Errorf("schtasks create: %v: %s", err, strings.TrimSpace(string(out)))
}
if out, err := exec.Command("schtasks", "/run", "/tn", updateTaskName).CombinedOutput(); err != nil {
return fmt.Errorf("schtasks run: %v: %s", err, strings.TrimSpace(string(out)))
}
return nil
}
func downloadFile(url, dest string) error {
resp, err := http.Get(url)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("HTTP %d from %s", resp.StatusCode, url)
}
f, err := os.Create(dest)
if err != nil {
return err
}
defer f.Close()
_, err = io.Copy(f, resp.Body)
return err
}
func httpGetBytes(url string) ([]byte, error) {
resp, err := http.Get(url)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("HTTP %d from %s", resp.StatusCode, url)
}
return io.ReadAll(resp.Body)
}
func verifyChecksum(filePath, filename string, checksumData []byte) error {
f, err := os.Open(filePath)
if err != nil {
return err
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return err
}
actual := hex.EncodeToString(h.Sum(nil))
for _, line := range strings.Split(string(checksumData), "\n") {
fields := strings.Fields(line)
if len(fields) == 2 && fields[1] == filename {
if fields[0] != actual {
return fmt.Errorf("expected %s got %s", fields[0], actual)
}
return nil
}
}
return fmt.Errorf("no checksum entry found for %s", filename)
}
func handleGenerateKey(cfg *config.Config, cmd *pb.ServerCommand) {
g := cmd.GenerateKey
label := g.Label
keyPath := fmt.Sprintf("/root/.ssh/vantage_%s", strings.ReplaceAll(label, " ", "_"))
opts := keys.KeyGenOptions{
KeyType: g.KeyType,
KeySize: g.KeySize,
Passphrase: g.Passphrase,
Comment: g.Comment,
}
pubKey, err := keys.GenerateKeyPair(keyPath, opts)
if err != nil {
log.Printf("key generation failed (cmd=%s): %v", cmd.CommandId, err)
return
}
privKeyData, err := os.ReadFile(keyPath)
if err != nil {
log.Printf("read private key failed (cmd=%s): %v", cmd.CommandId, err)
return
}
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
log.Printf("dial for key upload failed (cmd=%s): %v", cmd.CommandId, err)
return
}
defer client.Close()
keyID, err := client.UploadGeneratedKey(cfg.ServerID, cfg.AgentToken, pubKey, string(privKeyData), label)
if err != nil {
log.Printf("key upload failed (cmd=%s): %v", cmd.CommandId, err)
return
}
if err := keys.AddSSHIdentity(keyPath); err != nil {
log.Printf("add ssh identity failed (cmd=%s): %v", cmd.CommandId, err)
}
log.Printf("generated and uploaded key %q (key_id=%s, cmd=%s)", label, keyID, cmd.CommandId)
}
func localIP() string {
addrs, err := net.InterfaceAddrs()
if err != nil {
return ""
}
for _, addr := range addrs {
if ipNet, ok := addr.(*net.IPNet); ok && !ipNet.IP.IsLoopback() {
if ipNet.IP.To4() != nil {
return ipNet.IP.String()
}
}
}
return ""
}
func GenerateAndUpload(cfg *config.Config, label string) error {
client, err := grpcclient.New(cfg.ServerURL, cfg.TLS)
if err != nil {
return err
}
defer client.Close()
keyPath := fmt.Sprintf("/root/.ssh/vantage_%s", strings.ReplaceAll(label, " ", "_"))
pubKey, err := keys.GenerateKeyPair(keyPath, keys.KeyGenOptions{Comment: label})
if err != nil {
return err
}
privKeyData, err := os.ReadFile(keyPath)
if err != nil {
return fmt.Errorf("read private key: %w", err)
}
keyID, err := client.UploadGeneratedKey(cfg.ServerID, cfg.AgentToken, pubKey, string(privKeyData), label)
if err != nil {
return err
}
if err := keys.AddSSHIdentity(keyPath); err != nil {
log.Printf("add ssh identity: %v", err)
}
log.Printf("uploaded generated key %s (key_id=%s)", label, keyID)
return nil
}