Linux/Proxmox Agent + Plattform-Updater + Proxmox Frontend

- agent-linux/: Kompletter Linux/Proxmox RMM Agent
  - Collectors: CPU, Memory, Disk, Network, Services, Updates, Proxmox (VMs, Container, Storage, ZFS), Backups
  - Backup Collector: Jobs nach Node gefiltert (Cluster-aware)
  - WebSocket: Tunnel-Support, writeMux fuer stabile Verbindungen
  - Systemd Service Template

- updater/: Plattform-unabhaengiger Updater (FreeBSD + Linux)
  - runtime.GOOS erkennt Plattform automatisch
  - FreeBSD: /usr/local/rmm/, service rmm_agent
  - Linux: /usr/local/bin/, /etc/rmm/, systemctl rmm-agent
  - Firmware-Download mit ?platform= Parameter

- frontend/: Proxmox-Bereich
  - ProxmoxServers.jsx: Eigene Seite fuer PVE Server mit VM/CT Counts
  - ProxmoxPanel.jsx: Detail-Panel mit Uebersicht, VMs, Container, Storage, ZFS, Tunnel, Dienste, Updates, Backups
  - Agents.jsx: Filtert Linux-Agents raus (nur OPNsense)
  - Sidebar: Proxmox Navigationspunkt
  - Installationsanleitung mit rmm-agent-linux + rmm-updater-linux

- backend: Platform-Feld fuer Agents, Firmware multi-platform
This commit is contained in:
cynfo3000
2026-03-01 22:03:18 +01:00
parent c114df215a
commit 7f9fbd257f
39 changed files with 5775 additions and 112 deletions
+142
View File
@@ -0,0 +1,142 @@
# RMM Agent Linux
Linux/Proxmox Agent für das RMM-System. Sammelt Systemdaten und ermöglicht Remote-Management über WebSocket.
## Features
- **System-Monitoring**: CPU, Memory, Disk, Network, Services, Updates
- **Proxmox-Integration**: VMs, Container, Storage, Cluster, Subscription, ZFS Pools
- **Remote-Commands**: `exec` (beliebige Befehle), `reboot`
- **Agent-Updates**: Automatische Binary-Updates über WebSocket
- **Cross-Platform**: Funktioniert auf normalen Linux-Systemen und Proxmox VE
## Unterstützte Systeme
- Debian/Ubuntu (mit apt)
- Proxmox VE
- Andere Linux-Distributionen mit systemd
## Installation
1. **Binary kompilieren:**
```bash
make agent-linux VERSION=1.0.0
```
2. **Auf Zielsystem installieren:**
```bash
# Dateien kopieren
scp build/rmm-agent-linux root@target-host:/tmp/
scp agent-linux/install.sh root@target-host:/tmp/
scp agent-linux/rmm-agent-linux.service root@target-host:/tmp/
scp agent-linux/config.yaml.example root@target-host:/tmp/
# Installation
ssh root@target-host
cd /tmp
chmod +x install.sh
./install.sh
```
3. **Konfiguration anpassen:**
```bash
nano /etc/rmm/config.yaml
```
4. **Service starten:**
```bash
systemctl start rmm-agent-linux
systemctl status rmm-agent-linux
journalctl -u rmm-agent-linux -f
```
## Konfiguration
```yaml
backend_url: "https://your-rmm-backend.example.com"
api_key: "your-api-key-here"
interval_seconds: 60
agent_name: "linux-server-01"
insecure: true # Für Entwicklung, false für Produktion
```
## CLI-Flags
```bash
rmm-agent-linux [flags]
Flags:
-config string
Path zur Konfigurationsdatei (default "/etc/rmm/config.yaml")
-insecure
TLS-Zertifikatpruefung deaktivieren
-version
Version anzeigen
```
## Proxmox-Features
Wenn auf Proxmox VE installiert, sammelt der Agent zusätzlich:
- **VMs/Container**: Status, Ressourcen-Usage, Uptime
- **Storage**: Verfügbarer Platz, Typen (ZFS, LVM, NFS, etc.)
- **Cluster**: Node-Status, Quorum
- **Subscription**: Lizenz-Status
- **ZFS**: Pool-Health, Scrub-Status, Größen
## Systemd Service
- Service Name: `rmm-agent-linux`
- Config: `/etc/rmm/config.yaml`
- Agent ID: `/etc/rmm/agent_id` (automatisch generiert)
- Logs: `journalctl -u rmm-agent-linux`
## Remote-Commands
Über WebSocket unterstützte Commands:
- **exec**: Beliebige Shell-Befehle ausführen
- **reboot**: System-Neustart
- **agent_update**: Agent-Binary aktualisieren
## Entwicklung
```bash
# Dependencies installieren
cd agent-linux
go mod tidy
# Entwicklung
go run . -config config.yaml.example
# Cross-compile für Linux
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o rmm-agent-linux .
# Mit Version
go build -ldflags "-X main.Version=1.0.0" -o rmm-agent-linux .
```
## Unterschiede zum FreeBSD/OPNsense Agent
- **OS-Detection**: Linux-spezifische `/proc`-Dateien und `systemd`
- **Package-Updates**: `apt list --upgradable` statt `pkg`
- **Services**: `systemctl` statt `service`
- **Proxmox**: Zusätzliche `pvesh`-Integration für VMs/Container
- **Keine OPNsense-Features**: Kein WireGuard, DHCP, Certificates, etc.
## Troubleshooting
```bash
# Service Status
systemctl status rmm-agent-linux
# Live-Logs
journalctl -u rmm-agent-linux -f
# Config testen
rmm-agent-linux -config /etc/rmm/config.yaml -version
# Manuelle Registrierung testen
rm /etc/rmm/agent_id
systemctl restart rmm-agent-linux
```
+95
View File
@@ -0,0 +1,95 @@
package client
import (
"bytes"
"crypto/tls"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
)
type Client struct {
baseURL string
apiKey string
httpClient *http.Client
}
func New(baseURL, apiKey string, insecure bool) *Client {
transport := &http.Transport{
TLSClientConfig: &tls.Config{
InsecureSkipVerify: insecure,
},
}
return &Client{
baseURL: baseURL,
apiKey: apiKey,
httpClient: &http.Client{
Transport: transport,
Timeout: 30 * time.Second,
},
}
}
// Register - Agent am Backend registrieren, gibt Agent-ID zurueck
func (c *Client) Register(req interface{}) (string, error) {
body, err := json.Marshal(req)
if err != nil {
return "", err
}
resp, err := c.doRequest("POST", "/api/v1/agent/register", body)
if err != nil {
return "", err
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusCreated {
return "", fmt.Errorf("Registrierung fehlgeschlagen (HTTP %d): %s", resp.StatusCode, string(respBody))
}
var result struct {
ID string `json:"id"`
}
if err := json.Unmarshal(respBody, &result); err != nil {
return "", fmt.Errorf("Response parsen: %w", err)
}
return result.ID, nil
}
// Heartbeat - Systemdaten an Backend senden
func (c *Client) Heartbeat(req interface{}) error {
body, err := json.Marshal(req)
if err != nil {
return err
}
resp, err := c.doRequest("POST", "/api/v1/agent/heartbeat", body)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
respBody, _ := io.ReadAll(resp.Body)
return fmt.Errorf("Heartbeat fehlgeschlagen (HTTP %d): %s", resp.StatusCode, string(respBody))
}
return nil
}
func (c *Client) doRequest(method, path string, body []byte) (*http.Response, error) {
req, err := http.NewRequest(method, c.baseURL+path, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-API-Key", c.apiKey)
return c.httpClient.Do(req)
}
+89
View File
@@ -0,0 +1,89 @@
package collector
import (
"encoding/json"
"fmt"
"log"
"os"
"os/exec"
"strings"
"time"
)
type BackupTask struct {
UPID string `json:"upid"`
Status string `json:"status"`
StartTime int64 `json:"starttime"`
EndTime int64 `json:"endtime"`
Node string `json:"node"`
}
type BackupData struct {
Jobs []map[string]interface{} `json:"jobs"`
Tasks []BackupTask `json:"tasks"`
}
func CollectBackups() *BackupData {
if _, err := os.Stat("/usr/bin/pvesh"); err != nil {
return nil
}
result := &BackupData{
Jobs: []map[string]interface{}{},
Tasks: []BackupTask{},
}
// Hostname fuer Node-Filterung
hostname, _ := os.Hostname()
nodeName := strings.TrimSpace(hostname)
// Backup Jobs aus Cluster-Config — nur Jobs die auf diesem Node laufen
cmd := exec.Command("/usr/bin/pvesh", "get", "/cluster/backup", "--output-format", "json")
if out, err := cmd.Output(); err == nil {
var allJobs []map[string]interface{}
if err := json.Unmarshal(out, &allJobs); err == nil {
for _, j := range allJobs {
// next-run -> next_run umbenennen fuer Frontend-Kompatibilitaet
if nr, ok := j["next-run"]; ok {
j["next_run"] = nr
delete(j, "next-run")
}
// Filtern: Job gehoert zu diesem Node wenn:
// - node == unser hostname
// - node leer/nicht gesetzt (= alle Nodes)
jobNode, _ := j["node"].(string)
if jobNode == "" || strings.EqualFold(jobNode, nodeName) {
result.Jobs = append(result.Jobs, j)
}
}
log.Printf("Backups: %d/%d Jobs fuer Node %s", len(result.Jobs), len(allJobs), nodeName)
}
} else {
log.Printf("Backups: Jobs laden fehlgeschlagen: %v", err)
}
if nodeName == "" {
return result
}
since := fmt.Sprintf("%d", time.Now().AddDate(0, 0, -30).Unix())
cmd = exec.Command("/usr/bin/pvesh", "get", "/nodes/"+nodeName+"/tasks",
"--output-format", "json",
"--typefilter", "vzdump",
"--limit", "500",
"--since", since,
)
if out, err := cmd.Output(); err == nil {
var tasks []BackupTask
if err := json.Unmarshal(out, &tasks); err == nil {
result.Tasks = tasks
log.Printf("Backups: %d Tasks (letzte 30 Tage)", len(tasks))
} else {
log.Printf("Backups: Tasks JSON parse fehler: %v", err)
}
} else {
log.Printf("Backups: Tasks laden fehlgeschlagen: %v", err)
}
return result
}
+168
View File
@@ -0,0 +1,168 @@
package collector
import (
"os"
"strconv"
"strings"
"time"
)
type CPUInfo struct {
Model string
Cores int
Threads int
FreqMHz int
UsagePercent float64
}
func CollectCPU() CPUInfo {
cpu := CPUInfo{}
// CPU-Informationen aus /proc/cpuinfo lesen
if data, err := os.ReadFile("/proc/cpuinfo"); err == nil {
lines := strings.Split(string(data), "\n")
cpuCount := 0
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" {
continue
}
parts := strings.SplitN(line, ":", 2)
if len(parts) != 2 {
continue
}
key := strings.TrimSpace(parts[0])
value := strings.TrimSpace(parts[1])
switch key {
case "model name":
if cpu.Model == "" { // Nur vom ersten CPU nehmen
cpu.Model = value
}
case "cpu MHz":
if cpu.FreqMHz == 0 { // Nur vom ersten CPU nehmen
if freq, err := strconv.ParseFloat(value, 64); err == nil {
cpu.FreqMHz = int(freq)
}
}
case "processor":
cpuCount++
case "cpu cores":
if cores, err := strconv.Atoi(value); err == nil && cpu.Cores == 0 {
cpu.Cores = cores
}
case "siblings":
if threads, err := strconv.Atoi(value); err == nil && cpu.Threads == 0 {
cpu.Threads = threads
}
}
}
// Falls cpu cores und siblings nicht vorhanden, nutze processor count
if cpu.Cores == 0 {
cpu.Cores = cpuCount
}
if cpu.Threads == 0 {
cpu.Threads = cpuCount
}
}
// CPU-Usage berechnen
cpu.UsagePercent = calculateCPUUsage()
return cpu
}
// Berechnet die aktuelle CPU-Auslastung über zwei /proc/stat Messungen
func calculateCPUUsage() float64 {
stat1 := readCPUStat()
if stat1 == nil {
return 0.0
}
time.Sleep(100 * time.Millisecond) // Kurz warten für zweite Messung
stat2 := readCPUStat()
if stat2 == nil {
return 0.0
}
// Delta berechnen
totalDelta := (stat2.total - stat1.total)
idleDelta := (stat2.idle - stat1.idle)
if totalDelta <= 0 {
return 0.0
}
usage := 100.0 * float64(totalDelta-idleDelta) / float64(totalDelta)
if usage < 0 {
usage = 0
}
if usage > 100 {
usage = 100
}
return usage
}
type cpuStat struct {
total int64
idle int64
}
// Liest CPU-Statistiken aus /proc/stat
func readCPUStat() *cpuStat {
data, err := os.ReadFile("/proc/stat")
if err != nil {
return nil
}
lines := strings.Split(string(data), "\n")
if len(lines) == 0 {
return nil
}
// Erste Zeile sollte "cpu ..." sein
firstLine := strings.TrimSpace(lines[0])
if !strings.HasPrefix(firstLine, "cpu ") {
return nil
}
fields := strings.Fields(firstLine)
if len(fields) < 5 {
return nil
}
// cpu user nice system idle iowait irq softirq ...
var values []int64
for i := 1; i < len(fields) && i <= 7; i++ { // max 7 Werte lesen
if val, err := strconv.ParseInt(fields[i], 10, 64); err == nil {
values = append(values, val)
}
}
if len(values) < 4 {
return nil
}
// Total = Summe aller Werte
total := int64(0)
for _, val := range values {
total += val
}
// Idle = idle + iowait (falls iowait verfügbar)
idle := values[3] // idle
if len(values) > 4 {
idle += values[4] // + iowait
}
return &cpuStat{
total: total,
idle: idle,
}
}
+133
View File
@@ -0,0 +1,133 @@
package collector
import (
"os/exec"
"strconv"
"strings"
"syscall"
)
type DiskInfo struct {
Filesystem string `json:"filesystem"`
TotalBytes int64 `json:"total_bytes"`
UsedBytes int64 `json:"used_bytes"`
FreeBytes int64 `json:"free_bytes"`
MountPoint string `json:"mount_point"`
}
func CollectDisks() []DiskInfo {
var disks []DiskInfo
// df -T zur Anzeige aller mounted filesystems
output, err := exec.Command("/usr/bin/df", "-T", "-P").CombinedOutput()
if err != nil {
return disks
}
lines := strings.Split(string(output), "\n")
if len(lines) < 2 {
return disks
}
// Erste Zeile ist Header, überspringen
for _, line := range lines[1:] {
line = strings.TrimSpace(line)
if line == "" {
continue
}
fields := strings.Fields(line)
if len(fields) < 7 {
continue
}
filesystem := fields[0]
fstype := fields[1]
totalKB := fields[2]
usedKB := fields[3]
availKB := fields[4]
mountpoint := fields[6]
// Überspringe spezielle Filesysteme
if shouldSkipFilesystem(filesystem, fstype, mountpoint) {
continue
}
// Konvertiere kB zu Bytes
total, _ := strconv.ParseInt(totalKB, 10, 64)
used, _ := strconv.ParseInt(usedKB, 10, 64)
avail, _ := strconv.ParseInt(availKB, 10, 64)
// Für bessere Genauigkeit: Versuche statvfs syscall
if stat := getStatvfs(mountpoint); stat != nil {
total = int64(stat.Blocks) * int64(stat.Bsize)
avail = int64(stat.Bavail) * int64(stat.Bsize)
used = total - avail
} else {
// Fallback: df-Werte in Bytes konvertieren
total *= 1024
used *= 1024
avail *= 1024
}
disks = append(disks, DiskInfo{
Filesystem: filesystem,
TotalBytes: total,
UsedBytes: used,
FreeBytes: avail,
MountPoint: mountpoint,
})
}
return disks
}
// Bestimmt ob ein Filesystem übersprungen werden soll
func shouldSkipFilesystem(filesystem, fstype, mountpoint string) bool {
// Überspringe virtuelle Filesysteme
skipTypes := []string{
"proc", "sysfs", "devfs", "devtmpfs", "tmpfs", "debugfs",
"securityfs", "cgroup", "cgroup2", "pstore", "bpf",
"tracefs", "configfs", "fusectl", "binfmt_misc",
"autofs", "mqueue", "hugetlbfs", "sunrpc",
}
for _, skipType := range skipTypes {
if fstype == skipType {
return true
}
}
// Überspringe spezielle Mount-Points
skipMounts := []string{
"/dev", "/proc", "/sys", "/run", "/tmp",
"/dev/shm", "/run/lock", "/run/user",
}
for _, skipMount := range skipMounts {
if strings.HasPrefix(mountpoint, skipMount) {
return true
}
}
// Überspringe Loop-Devices außer root
if strings.HasPrefix(filesystem, "/dev/loop") && mountpoint != "/" {
return true
}
// Überspringe Bind-Mounts (heuristic)
if strings.Contains(mountpoint, "/snap/") {
return true
}
return false
}
// Holt präzise Filesystem-Statistiken via statvfs syscall
func getStatvfs(path string) *syscall.Statfs_t {
var stat syscall.Statfs_t
if err := syscall.Statfs(path, &stat); err != nil {
return nil
}
return &stat
}
+70
View File
@@ -0,0 +1,70 @@
package collector
import (
"os"
"strconv"
"strings"
)
type MemoryInfo struct {
TotalBytes int64
UsedBytes int64
FreeBytes int64
}
func CollectMemory() MemoryInfo {
mem := MemoryInfo{}
data, err := os.ReadFile("/proc/meminfo")
if err != nil {
return mem
}
lines := strings.Split(string(data), "\n")
values := make(map[string]int64)
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" {
continue
}
parts := strings.Fields(line)
if len(parts) < 2 {
continue
}
key := strings.TrimSuffix(parts[0], ":")
valueStr := parts[1]
// Wert in kB, konvertiere zu Bytes
if value, err := strconv.ParseInt(valueStr, 10, 64); err == nil {
values[key] = value * 1024 // kB zu Bytes
}
}
// Basis-Werte
memTotal := values["MemTotal"]
memFree := values["MemFree"]
memAvailable := values["MemAvailable"]
buffers := values["Buffers"]
cached := values["Cached"]
mem.TotalBytes = memTotal
// Verfügbarer Speicher: Nutze MemAvailable falls vorhanden, sonst berechne
if memAvailable > 0 {
mem.FreeBytes = memAvailable
} else {
// Fallback: Free + Buffers + Cached (grobe Schätzung)
mem.FreeBytes = memFree + buffers + cached
}
// Benutzter Speicher = Total - Free
if mem.FreeBytes > mem.TotalBytes {
mem.FreeBytes = mem.TotalBytes
}
mem.UsedBytes = mem.TotalBytes - mem.FreeBytes
return mem
}
+185
View File
@@ -0,0 +1,185 @@
package collector
import (
"os"
"os/exec"
"strconv"
"strings"
)
type NetworkInterface struct {
Name string `json:"name"`
Role string `json:"role"`
IP string `json:"ip"`
MAC string `json:"mac"`
Status string `json:"status"`
RxBytes int64 `json:"rx_bytes"`
TxBytes int64 `json:"tx_bytes"`
}
func CollectNetwork() []NetworkInterface {
var interfaces []NetworkInterface
// Interface-Statistiken aus /proc/net/dev lesen
stats := readNetDevStats()
// Interface-Informationen aus /sys/class/net/* sammeln
entries, err := os.ReadDir("/sys/class/net")
if err != nil {
return interfaces
}
for _, entry := range entries {
if !entry.IsDir() {
continue
}
ifName := entry.Name()
// Überspringe loopback
if ifName == "lo" {
continue
}
iface := NetworkInterface{
Name: ifName,
Role: determineInterfaceRole(ifName),
}
// Status aus operstate lesen
if data, err := os.ReadFile("/sys/class/net/" + ifName + "/operstate"); err == nil {
state := strings.TrimSpace(string(data))
if state == "up" {
iface.Status = "up"
} else {
iface.Status = "down"
}
}
// MAC-Adresse lesen
if data, err := os.ReadFile("/sys/class/net/" + ifName + "/address"); err == nil {
iface.MAC = strings.TrimSpace(string(data))
}
// Statistiken hinzufügen
if stat, exists := stats[ifName]; exists {
iface.RxBytes = stat.rxBytes
iface.TxBytes = stat.txBytes
}
// IP-Adresse über ip addr bestimmen
iface.IP = getInterfaceIP(ifName)
interfaces = append(interfaces, iface)
}
return interfaces
}
// Bestimmt die Rolle eines Interfaces basierend auf dem Namen
func determineInterfaceRole(ifName string) string {
switch {
case strings.HasPrefix(ifName, "eth"):
return "lan"
case strings.HasPrefix(ifName, "ens"):
return "lan"
case strings.HasPrefix(ifName, "enp"):
return "lan"
case strings.HasPrefix(ifName, "wlan"):
return "wan"
case strings.HasPrefix(ifName, "wlp"):
return "wan"
case strings.HasPrefix(ifName, "ppp"):
return "wan"
case strings.HasPrefix(ifName, "tun"):
return "vpn"
case strings.HasPrefix(ifName, "tap"):
return "vpn"
case strings.HasPrefix(ifName, "br"):
return "bridge"
case strings.HasPrefix(ifName, "docker"):
return "virtual"
case strings.HasPrefix(ifName, "veth"):
return "virtual"
case strings.HasPrefix(ifName, "virbr"):
return "virtual"
default:
return "other"
}
}
// Liest die erste IPv4-Adresse eines Interfaces
func getInterfaceIP(ifName string) string {
output, err := exec.Command("/usr/sbin/ip", "-4", "addr", "show", ifName).CombinedOutput()
if err != nil {
return ""
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "inet ") {
fields := strings.Fields(line)
if len(fields) >= 2 {
// inet 192.168.1.100/24 ... -> 192.168.1.100
ipCidr := fields[1]
if parts := strings.SplitN(ipCidr, "/", 2); len(parts) > 0 {
return parts[0]
}
}
}
}
return ""
}
// Network-Statistiken
type netDevStat struct {
rxBytes int64
txBytes int64
}
// Liest Interface-Statistiken aus /proc/net/dev
func readNetDevStats() map[string]netDevStat {
stats := make(map[string]netDevStat)
data, err := os.ReadFile("/proc/net/dev")
if err != nil {
return stats
}
lines := strings.Split(string(data), "\n")
if len(lines) < 3 {
return stats
}
// Ersten 2 Zeilen sind Header
for _, line := range lines[2:] {
line = strings.TrimSpace(line)
if line == "" {
continue
}
// Interface: rx_bytes rx_packets ... tx_bytes tx_packets ...
parts := strings.SplitN(line, ":", 2)
if len(parts) != 2 {
continue
}
ifName := strings.TrimSpace(parts[0])
values := strings.Fields(strings.TrimSpace(parts[1]))
if len(values) >= 9 {
// rx_bytes ist das erste Feld, tx_bytes das 9te Feld (0-basiert: Index 8)
rxBytes, _ := strconv.ParseInt(values[0], 10, 64)
txBytes, _ := strconv.ParseInt(values[8], 10, 64)
stats[ifName] = netDevStat{
rxBytes: rxBytes,
txBytes: txBytes,
}
}
}
return stats
}
+506
View File
@@ -0,0 +1,506 @@
package collector
import (
"encoding/json"
"log"
"os"
"os/exec"
"strconv"
"strings"
)
type ProxmoxInfo struct {
Available bool `json:"available"`
Version string `json:"version"`
Node string `json:"node"`
VMs []ProxmoxVM `json:"vms,omitempty"`
Containers []ProxmoxContainer `json:"containers,omitempty"`
Storage []ProxmoxStorage `json:"storage,omitempty"`
Cluster *ProxmoxCluster `json:"cluster,omitempty"`
Subscription *ProxmoxSubscription `json:"subscription,omitempty"`
ZFSPools []ZFSPool `json:"zfs_pools,omitempty"`
}
type ProxmoxVM struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
Type string `json:"type"`
CPUUsage *float64 `json:"cpu_usage,omitempty"`
MemoryUsed *int64 `json:"memory_used,omitempty"`
MemoryMax *int64 `json:"memory_max,omitempty"`
DiskUsed *int64 `json:"disk_used,omitempty"`
DiskMax *int64 `json:"disk_max,omitempty"`
Uptime *int64 `json:"uptime,omitempty"`
Node string `json:"node"`
}
type ProxmoxContainer struct {
VMID int `json:"vmid"`
Name string `json:"name"`
Status string `json:"status"`
Type string `json:"type"`
CPUUsage *float64 `json:"cpu_usage,omitempty"`
MemoryUsed *int64 `json:"memory_used,omitempty"`
MemoryMax *int64 `json:"memory_max,omitempty"`
DiskUsed *int64 `json:"disk_used,omitempty"`
DiskMax *int64 `json:"disk_max,omitempty"`
Uptime *int64 `json:"uptime,omitempty"`
Node string `json:"node"`
}
type ProxmoxStorage struct {
Name string `json:"storage"`
Type string `json:"type"`
Total int64 `json:"total"`
Used int64 `json:"used"`
Available int64 `json:"available"`
Enabled bool `json:"enabled"`
Active bool `json:"active"`
Content string `json:"content"`
}
type ProxmoxCluster struct {
Name string `json:"name"`
Nodes []ProxmoxClusterNode `json:"nodes"`
Quorum bool `json:"quorum"`
}
type ProxmoxClusterNode struct {
Name string `json:"name"`
Status string `json:"status"`
Local bool `json:"local"`
}
type ProxmoxSubscription struct {
Status string `json:"status"`
ProductName string `json:"productname,omitempty"`
Key string `json:"key,omitempty"`
NextDueDate string `json:"next_due_date,omitempty"`
}
type ZFSPool struct {
Name string `json:"name"`
Size int64 `json:"size"`
Allocated int64 `json:"allocated"`
Free int64 `json:"free"`
Health string `json:"health"`
Fragmentation string `json:"fragmentation,omitempty"`
ScrubStatus string `json:"scrub_status,omitempty"`
}
func CollectProxmox() *ProxmoxInfo {
proxmox := &ProxmoxInfo{
Available: false,
}
// Prüfen ob pvesh verfügbar ist
// Voller Pfad fuer pvesh, da systemd reduzierten PATH hat
if _, err := os.Stat("/usr/bin/pvesh"); err != nil {
log.Printf("Proxmox: pvesh nicht gefunden: %v", err)
return nil // Proxmox nicht verfügbar
}
log.Printf("Proxmox: pvesh gefunden, sammle Daten...")
proxmox.Available = true
// Version abrufen
if data := pveshGet("/version"); data != nil {
log.Printf("Proxmox: Version data: %v", data)
if versionData, ok := data.(map[string]interface{}); ok {
if version, ok := versionData["version"].(string); ok {
proxmox.Version = version
}
}
}
// Node-Name bestimmen
proxmox.Node = getLocalNodeName()
// VMs sammeln
proxmox.VMs = collectProxmoxVMs()
// Container sammeln
proxmox.Containers = collectProxmoxContainers()
// Storage sammeln
proxmox.Storage = collectProxmoxStorage()
// Cluster-Info sammeln
proxmox.Cluster = collectProxmoxCluster()
// Subscription-Info sammeln
proxmox.Subscription = collectProxmoxSubscription()
// ZFS-Pools sammeln
proxmox.ZFSPools = collectZFSPools()
return proxmox
}
// pvesh get wrapper
func pveshGet(path string) interface{} {
cmd := exec.Command("/usr/bin/pvesh", "get", path, "--output-format", "json")
output, err := cmd.Output()
if err != nil {
log.Printf("Proxmox: pvesh get %s fehler: %v", path, err)
return nil
}
log.Printf("Proxmox: pvesh get %s: %d bytes", path, len(output))
var result interface{}
if err := json.Unmarshal(output, &result); err != nil {
log.Printf("Proxmox: JSON parse fehler fuer %s: %v", path, err)
return nil
}
return result
}
func getLocalNodeName() string {
// Hostname als Node-Name verwenden
if hostname, err := os.Hostname(); err == nil {
return hostname
}
return "localhost"
}
func collectProxmoxVMs() []ProxmoxVM {
var vms []ProxmoxVM
nodeName := getLocalNodeName()
data := pveshGet("/nodes/" + nodeName + "/qemu")
if data == nil {
return vms
}
if vmList, ok := data.([]interface{}); ok {
for _, item := range vmList {
if vmData, ok := item.(map[string]interface{}); ok {
vm := ProxmoxVM{
Type: "qemu",
Node: nodeName,
}
if vmid, ok := vmData["vmid"].(float64); ok {
vm.VMID = int(vmid)
}
if name, ok := vmData["name"].(string); ok {
vm.Name = name
}
if status, ok := vmData["status"].(string); ok {
vm.Status = status
}
if cpu, ok := vmData["cpu"].(float64); ok {
vm.CPUUsage = &cpu
}
if mem, ok := vmData["mem"].(float64); ok {
memUsed := int64(mem)
vm.MemoryUsed = &memUsed
}
if maxmem, ok := vmData["maxmem"].(float64); ok {
maxMem := int64(maxmem)
vm.MemoryMax = &maxMem
}
if disk, ok := vmData["disk"].(float64); ok {
diskUsed := int64(disk)
vm.DiskUsed = &diskUsed
}
if maxdisk, ok := vmData["maxdisk"].(float64); ok {
maxDisk := int64(maxdisk)
vm.DiskMax = &maxDisk
}
if uptime, ok := vmData["uptime"].(float64); ok {
uptimeVal := int64(uptime)
vm.Uptime = &uptimeVal
}
vms = append(vms, vm)
}
}
}
return vms
}
func collectProxmoxContainers() []ProxmoxContainer {
var containers []ProxmoxContainer
nodeName := getLocalNodeName()
data := pveshGet("/nodes/" + nodeName + "/lxc")
if data == nil {
return containers
}
if containerList, ok := data.([]interface{}); ok {
for _, item := range containerList {
if containerData, ok := item.(map[string]interface{}); ok {
container := ProxmoxContainer{
Type: "lxc",
Node: nodeName,
}
if vmid, ok := containerData["vmid"].(float64); ok {
container.VMID = int(vmid)
}
if name, ok := containerData["name"].(string); ok {
container.Name = name
}
if status, ok := containerData["status"].(string); ok {
container.Status = status
}
if cpu, ok := containerData["cpu"].(float64); ok {
container.CPUUsage = &cpu
}
if mem, ok := containerData["mem"].(float64); ok {
memUsed := int64(mem)
container.MemoryUsed = &memUsed
}
if maxmem, ok := containerData["maxmem"].(float64); ok {
maxMem := int64(maxmem)
container.MemoryMax = &maxMem
}
if disk, ok := containerData["disk"].(float64); ok {
diskUsed := int64(disk)
container.DiskUsed = &diskUsed
}
if maxdisk, ok := containerData["maxdisk"].(float64); ok {
maxDisk := int64(maxdisk)
container.DiskMax = &maxDisk
}
if uptime, ok := containerData["uptime"].(float64); ok {
uptimeVal := int64(uptime)
container.Uptime = &uptimeVal
}
containers = append(containers, container)
}
}
}
return containers
}
func collectProxmoxStorage() []ProxmoxStorage {
var storages []ProxmoxStorage
nodeName := getLocalNodeName()
data := pveshGet("/nodes/" + nodeName + "/storage")
if data == nil {
return storages
}
if storageList, ok := data.([]interface{}); ok {
for _, item := range storageList {
if storageData, ok := item.(map[string]interface{}); ok {
storage := ProxmoxStorage{}
if name, ok := storageData["storage"].(string); ok {
storage.Name = name
}
if storageType, ok := storageData["type"].(string); ok {
storage.Type = storageType
}
if total, ok := storageData["total"].(float64); ok {
storage.Total = int64(total)
}
if used, ok := storageData["used"].(float64); ok {
storage.Used = int64(used)
}
if avail, ok := storageData["avail"].(float64); ok {
storage.Available = int64(avail)
}
if enabled, ok := storageData["enabled"].(float64); ok {
storage.Enabled = enabled > 0
}
if active, ok := storageData["active"].(float64); ok {
storage.Active = active > 0
}
if content, ok := storageData["content"].(string); ok {
storage.Content = content
}
storages = append(storages, storage)
}
}
}
return storages
}
func collectProxmoxCluster() *ProxmoxCluster {
data := pveshGet("/cluster/status")
if data == nil {
return nil
}
cluster := &ProxmoxCluster{}
if clusterData, ok := data.([]interface{}); ok {
for _, item := range clusterData {
if itemData, ok := item.(map[string]interface{}); ok {
if itemType, ok := itemData["type"].(string); ok {
if itemType == "cluster" {
if name, ok := itemData["name"].(string); ok {
cluster.Name = name
}
if quorum, ok := itemData["quorate"].(float64); ok {
cluster.Quorum = quorum > 0
}
} else if itemType == "node" {
node := ProxmoxClusterNode{}
if name, ok := itemData["name"].(string); ok {
node.Name = name
}
if online, ok := itemData["online"].(float64); ok {
if online > 0 {
node.Status = "online"
} else {
node.Status = "offline"
}
}
if local, ok := itemData["local"].(float64); ok {
node.Local = local > 0
}
cluster.Nodes = append(cluster.Nodes, node)
}
}
}
}
}
if cluster.Name == "" && len(cluster.Nodes) == 0 {
return nil // Kein Cluster
}
return cluster
}
func collectProxmoxSubscription() *ProxmoxSubscription {
nodeName := getLocalNodeName()
data := pveshGet("/nodes/" + nodeName + "/subscription")
if data == nil {
return nil
}
subscription := &ProxmoxSubscription{}
if subData, ok := data.(map[string]interface{}); ok {
if status, ok := subData["status"].(string); ok {
subscription.Status = status
}
if productName, ok := subData["productname"].(string); ok {
subscription.ProductName = productName
}
if key, ok := subData["key"].(string); ok {
subscription.Key = key
}
if nextDueDate, ok := subData["nextduedate"].(string); ok {
subscription.NextDueDate = nextDueDate
}
}
return subscription
}
func collectZFSPools() []ZFSPool {
var pools []ZFSPool
// zpool list -H -o name,size,alloc,free,health
cmd := exec.Command("/usr/sbin/zpool", "list", "-H", "-o", "name,size,alloc,free,health")
output, err := cmd.Output()
if err != nil {
return pools // ZFS nicht verfügbar oder keine Pools
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" {
continue
}
fields := strings.Fields(line)
if len(fields) < 5 {
continue
}
pool := ZFSPool{
Name: fields[0],
Health: fields[4],
}
// Größen parsen (können Suffixe wie K, M, G, T haben)
pool.Size = parseZFSSize(fields[1])
pool.Allocated = parseZFSSize(fields[2])
pool.Free = parseZFSSize(fields[3])
// Zusätzliche Informationen aus zpool status
pool.ScrubStatus = getZFSPoolScrubStatus(pool.Name)
pools = append(pools, pool)
}
return pools
}
// Parst ZFS-Größen mit Einheiten (z.B. "1.2T" -> Bytes)
func parseZFSSize(sizeStr string) int64 {
if sizeStr == "-" || sizeStr == "" {
return 0
}
sizeStr = strings.TrimSpace(sizeStr)
if len(sizeStr) == 0 {
return 0
}
// Letzes Zeichen prüfen für Einheit
lastChar := sizeStr[len(sizeStr)-1]
var multiplier int64 = 1
var numberStr string
switch lastChar {
case 'K':
multiplier = 1024
numberStr = sizeStr[:len(sizeStr)-1]
case 'M':
multiplier = 1024 * 1024
numberStr = sizeStr[:len(sizeStr)-1]
case 'G':
multiplier = 1024 * 1024 * 1024
numberStr = sizeStr[:len(sizeStr)-1]
case 'T':
multiplier = 1024 * 1024 * 1024 * 1024
numberStr = sizeStr[:len(sizeStr)-1]
default:
numberStr = sizeStr
}
if value, err := strconv.ParseFloat(numberStr, 64); err == nil {
return int64(value * float64(multiplier))
}
return 0
}
// Holt Scrub-Status für einen ZFS-Pool
func getZFSPoolScrubStatus(poolName string) string {
cmd := exec.Command("/usr/sbin/zpool", "status", poolName)
output, err := cmd.Output()
if err != nil {
return ""
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if strings.Contains(line, "scrub") {
return line
}
}
return ""
}
+94
View File
@@ -0,0 +1,94 @@
package collector
import (
"os/exec"
"strings"
)
type ServiceInfo struct {
Name string `json:"name"`
Description string `json:"description"`
Status string `json:"status"`
}
func CollectServices() []ServiceInfo {
var services []ServiceInfo
// systemctl list-units --type=service --all --no-pager --no-legend
output, err := exec.Command("/usr/bin/systemctl", "list-units", "--type=service", "--all", "--no-pager", "--no-legend").CombinedOutput()
if err != nil {
return services
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" {
continue
}
// Format: UNIT LOAD ACTIVE SUB DESCRIPTION
fields := strings.Fields(line)
if len(fields) < 4 {
continue
}
serviceName := fields[0]
load := fields[1]
active := fields[2]
sub := fields[3]
// Beschreibung ist der Rest ab Feld 4
description := ""
if len(fields) > 4 {
description = strings.Join(fields[4:], " ")
}
// Status bestimmen
status := determineServiceStatus(load, active, sub)
// .service suffix entfernen für bessere Lesbarkeit
displayName := strings.TrimSuffix(serviceName, ".service")
services = append(services, ServiceInfo{
Name: displayName,
Description: description,
Status: status,
})
}
return services
}
// Bestimmt den Service-Status basierend auf systemctl Ausgabe
func determineServiceStatus(load, active, sub string) string {
// Priorität auf active state
switch active {
case "active":
// Wenn active, schaue auf sub-state
switch sub {
case "running":
return "running"
case "exited":
return "stopped" // one-shot service, beendet
default:
return "active"
}
case "inactive":
return "stopped"
case "failed":
return "error"
case "activating":
return "starting"
case "deactivating":
return "stopping"
default:
// Falls load state problematisch ist
if load == "not-found" {
return "not-found"
} else if load == "error" {
return "error"
}
return "unknown"
}
}
+132
View File
@@ -0,0 +1,132 @@
package collector
import (
"os"
"os/exec"
"strconv"
"strings"
"time"
)
// Basis-Systemdaten sammeln
func CollectHostname() string {
if data, err := os.ReadFile("/etc/hostname"); err == nil {
return strings.TrimSpace(string(data))
}
// Fallback
if hostname, err := os.Hostname(); err == nil {
return hostname
}
return "unknown"
}
func CollectOSVersion() string {
// Versuche verschiedene Wege die OS-Version zu bestimmen
// 1. /etc/os-release (Standard für moderne Linux-Distributionen)
if data, err := os.ReadFile("/etc/os-release"); err == nil {
lines := strings.Split(string(data), "\n")
var name, version string
for _, line := range lines {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "NAME=") {
name = strings.Trim(strings.SplitN(line, "=", 2)[1], "\"")
} else if strings.HasPrefix(line, "VERSION=") {
version = strings.Trim(strings.SplitN(line, "=", 2)[1], "\"")
}
}
if name != "" {
if version != "" {
return name + " " + version
}
return name
}
}
// 2. /etc/debian_version (Debian/Ubuntu)
if data, err := os.ReadFile("/etc/debian_version"); err == nil {
version := strings.TrimSpace(string(data))
return "Debian " + version
}
// 3. /etc/redhat-release (RHEL/CentOS)
if data, err := os.ReadFile("/etc/redhat-release"); err == nil {
return strings.TrimSpace(string(data))
}
// 4. uname fallback
if output, err := exec.Command("/usr/bin/uname", "-sr").CombinedOutput(); err == nil {
return strings.TrimSpace(string(output))
}
return "Linux (unknown)"
}
func CollectUptime() int64 {
if data, err := os.ReadFile("/proc/uptime"); err == nil {
parts := strings.Fields(string(data))
if len(parts) > 0 {
if uptime, err := strconv.ParseFloat(parts[0], 64); err == nil {
return int64(uptime)
}
}
}
// Fallback: Boot-Zeit berechnen
if stat, err := os.Stat("/proc/1"); err == nil {
bootTime := stat.ModTime()
return int64(time.Since(bootTime).Seconds())
}
return 0
}
// Hardware-Informationen sammeln
type HardwareInfo struct {
Manufacturer string
Model string
Serial string
BIOSVersion string
}
func CollectHardware() HardwareInfo {
hw := HardwareInfo{}
// DMI-Informationen aus /sys/class/dmi/id/ lesen
if data, err := os.ReadFile("/sys/class/dmi/id/sys_vendor"); err == nil {
hw.Manufacturer = strings.TrimSpace(string(data))
}
if data, err := os.ReadFile("/sys/class/dmi/id/product_name"); err == nil {
hw.Model = strings.TrimSpace(string(data))
}
if data, err := os.ReadFile("/sys/class/dmi/id/product_serial"); err == nil {
hw.Serial = strings.TrimSpace(string(data))
}
if data, err := os.ReadFile("/sys/class/dmi/id/bios_version"); err == nil {
hw.BIOSVersion = strings.TrimSpace(string(data))
}
// Fallback für Manufacturer
if hw.Manufacturer == "" {
if data, err := os.ReadFile("/sys/class/dmi/id/board_vendor"); err == nil {
hw.Manufacturer = strings.TrimSpace(string(data))
}
}
// Fallback für Model
if hw.Model == "" {
if data, err := os.ReadFile("/sys/class/dmi/id/board_name"); err == nil {
hw.Model = strings.TrimSpace(string(data))
}
}
return hw
}
+162
View File
@@ -0,0 +1,162 @@
package collector
import (
"os/exec"
"strings"
)
type UpdateInfo struct {
UpdateAvailable bool `json:"update_available"`
PendingCount int `json:"pending_count"`
Updates []PendingUpdateInfo `json:"updates"`
}
type PendingUpdateInfo struct {
Package string `json:"package"`
CurrentVer string `json:"current_version"`
NewVer string `json:"new_version"`
}
func CollectUpdates() *UpdateInfo {
updates := &UpdateInfo{
UpdateAvailable: false,
PendingCount: 0,
Updates: []PendingUpdateInfo{},
}
// apt list --upgradable 2>/dev/null
cmd := exec.Command("/usr/bin/apt", "list", "--upgradable")
output, err := cmd.CombinedOutput()
if err != nil {
// Fallback: apt-check falls verfügbar
return collectUpdatesAptCheck()
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "WARNING:") || strings.HasPrefix(line, "Listing...") {
continue
}
// Format: package/suite version [upgradable from: old-version]
// Beispiel: curl/focal-updates 7.68.0-1ubuntu2.7 amd64 [upgradable from: 7.68.0-1ubuntu2.6]
if !strings.Contains(line, "[upgradable from:") {
continue
}
// Package-Name extrahieren (bis zum ersten /)
parts := strings.SplitN(line, "/", 2)
if len(parts) < 2 {
continue
}
packageName := parts[0]
// Neue Version extrahieren (zwischen / und [upgradable from:)
remaining := parts[1]
beforeUpgradable := strings.Split(remaining, " [upgradable from:")
if len(beforeUpgradable) < 2 {
continue
}
// Neue Version ist im ersten Teil, aber wir müssen die Architektur entfernen
versionParts := strings.Fields(beforeUpgradable[0])
if len(versionParts) < 1 {
continue
}
newVersion := versionParts[0]
// Alte Version extrahieren (zwischen "from:" und "]")
oldVersionPart := beforeUpgradable[1]
oldVersion := strings.TrimSuffix(strings.TrimSpace(oldVersionPart), "]")
updates.Updates = append(updates.Updates, PendingUpdateInfo{
Package: packageName,
CurrentVer: oldVersion,
NewVer: newVersion,
})
}
updates.PendingCount = len(updates.Updates)
updates.UpdateAvailable = updates.PendingCount > 0
return updates
}
// Fallback-Methode mit apt-check (Ubuntu/Debian)
func collectUpdatesAptCheck() *UpdateInfo {
updates := &UpdateInfo{
UpdateAvailable: false,
PendingCount: 0,
Updates: []PendingUpdateInfo{},
}
// /usr/lib/update-notifier/apt-check
cmd := exec.Command("/usr/lib/update-notifier/apt-check")
output, err := cmd.CombinedOutput()
if err != nil {
// Noch ein Fallback: apt-get -s upgrade
return collectUpdatesAptGetSimulate()
}
// apt-check gibt "updates;security-updates" auf stderr aus
result := strings.TrimSpace(string(output))
parts := strings.SplitN(result, ";", 2)
if len(parts) >= 1 && parts[0] != "0" {
// Packages verfügbar, aber keine Details ohne apt list
updates.UpdateAvailable = true
updates.PendingCount = 1 // Platzhalter
updates.Updates = []PendingUpdateInfo{
{
Package: "system-updates",
CurrentVer: "various",
NewVer: "available",
},
}
}
return updates
}
// Letzter Fallback mit apt-get simulate
func collectUpdatesAptGetSimulate() *UpdateInfo {
updates := &UpdateInfo{
UpdateAvailable: false,
PendingCount: 0,
Updates: []PendingUpdateInfo{},
}
// apt-get -s upgrade (simulate)
cmd := exec.Command("/usr/bin/apt-get", "-s", "upgrade")
output, err := cmd.CombinedOutput()
if err != nil {
return updates
}
lines := strings.Split(string(output), "\n")
for _, line := range lines {
line = strings.TrimSpace(line)
// Suche nach "Inst package-name" Zeilen
if strings.HasPrefix(line, "Inst ") {
fields := strings.Fields(line)
if len(fields) >= 3 {
packageName := fields[1]
// Sehr simple Erkennung, bessere Parsing schwierig ohne apt list
updates.Updates = append(updates.Updates, PendingUpdateInfo{
Package: packageName,
CurrentVer: "unknown",
NewVer: "available",
})
}
}
}
updates.PendingCount = len(updates.Updates)
updates.UpdateAvailable = updates.PendingCount > 0
return updates
}
+8
View File
@@ -0,0 +1,8 @@
# RMM Agent Linux Configuration
# Copy to /etc/rmm/config.yaml and adjust values
backend_url: "https://your-rmm-backend.example.com"
api_key: "your-api-key-here"
interval_seconds: 60
agent_name: "linux-server-01"
insecure: true # Set to false in production with valid TLS certificates
+33
View File
@@ -0,0 +1,33 @@
package config
import (
"os"
"gopkg.in/yaml.v3"
)
type Config struct {
BackendURL string `yaml:"backend_url"`
APIKey string `yaml:"api_key"`
IntervalSeconds int `yaml:"interval_seconds"`
AgentName string `yaml:"agent_name"`
Insecure bool `yaml:"insecure"`
}
func Load(path string) (*Config, error) {
cfg := &Config{
IntervalSeconds: 60,
Insecure: true,
}
data, err := os.ReadFile(path)
if err != nil {
return nil, err
}
if err := yaml.Unmarshal(data, cfg); err != nil {
return nil, err
}
return cfg, nil
}
+8
View File
@@ -0,0 +1,8 @@
module github.com/cynfo/rmm-agent-linux
go 1.21
require (
github.com/gorilla/websocket v1.5.0
gopkg.in/yaml.v3 v3.0.1
)
+6
View File
@@ -0,0 +1,6 @@
github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+46
View File
@@ -0,0 +1,46 @@
#!/bin/bash
# RMM Agent Linux Installation Script
set -e
echo "==> Installing RMM Agent Linux"
# Check if running as root
if [[ $EUID -ne 0 ]]; then
echo "This script must be run as root"
exit 1
fi
# Create directories
mkdir -p /etc/rmm
mkdir -p /usr/local/bin
# Copy binary
if [ ! -f "rmm-agent-linux" ]; then
echo "Error: rmm-agent-linux binary not found in current directory"
exit 1
fi
cp rmm-agent-linux /usr/local/bin/rmm-agent-linux
chmod +x /usr/local/bin/rmm-agent-linux
# Copy service file
cp rmm-agent-linux.service /etc/systemd/system/
# Create example config if it doesn't exist
if [ ! -f "/etc/rmm/config.yaml" ]; then
cp config.yaml.example /etc/rmm/config.yaml
echo "==> Created /etc/rmm/config.yaml - please edit with your backend URL and API key"
fi
# Reload systemd and enable service
systemctl daemon-reload
systemctl enable rmm-agent-linux
echo "==> Installation complete!"
echo ""
echo "Next steps:"
echo "1. Edit /etc/rmm/config.yaml with your backend URL and API key"
echo "2. Start the service: systemctl start rmm-agent-linux"
echo "3. Check status: systemctl status rmm-agent-linux"
echo "4. View logs: journalctl -u rmm-agent-linux -f"
+203
View File
@@ -0,0 +1,203 @@
package main
import (
"flag"
"log"
"math"
"os"
"path/filepath"
"strings"
"time"
"github.com/cynfo/rmm-agent-linux/client"
"github.com/cynfo/rmm-agent-linux/collector"
"github.com/cynfo/rmm-agent-linux/config"
"github.com/cynfo/rmm-agent-linux/ws"
)
var Version = "dev"
func main() {
cfgPath := flag.String("config", "/etc/rmm/config.yaml", "Path zur Konfigurationsdatei")
insecure := flag.Bool("insecure", false, "TLS-Zertifikatpruefung deaktivieren")
version := flag.Bool("version", false, "Version anzeigen")
flag.Parse()
if *version {
log.Printf("RMM Agent Linux v%s", Version)
return
}
cfg, err := config.Load(*cfgPath)
if err != nil {
log.Fatalf("Konfiguration laden fehlgeschlagen: %v", err)
}
if *insecure {
cfg.Insecure = true
}
log.Printf("RMM Agent Linux v%s startet: %s", Version, cfg.AgentName)
log.Printf("Backend: %s, Intervall: %ds", cfg.BackendURL, cfg.IntervalSeconds)
c := client.New(cfg.BackendURL, cfg.APIKey, cfg.Insecure)
// Agent-ID persistent laden oder neu registrieren
idFile := "/etc/rmm/agent_id"
var agentID string
if data, err := os.ReadFile(idFile); err == nil && len(strings.TrimSpace(string(data))) > 0 {
agentID = strings.TrimSpace(string(data))
log.Printf("Agent-ID aus Datei geladen: %s", agentID)
}
// Registrierung mit Retry (sendet bestehende ID mit, falls vorhanden)
backoff := 5 * time.Second
for {
hostname := collector.CollectHostname()
if hostname == "" {
hostname, _ = os.Hostname()
}
// Bestimme OS-Version und Platform
osVersion := collector.CollectOSVersion()
platform := "linux"
regReq := map[string]string{
"name": cfg.AgentName,
"hostname": hostname,
"ip": getLocalIP(),
"opnsense_version": osVersion, // Verwende OS-Version statt OPNsense
"agent_version": Version,
"platform": platform,
}
if agentID != "" {
regReq["agent_id"] = agentID
}
newID, regErr := c.Register(regReq)
if regErr != nil {
log.Printf("Registrierung fehlgeschlagen: %v (Retry in %v)", regErr, backoff)
time.Sleep(backoff)
backoff = time.Duration(math.Min(float64(backoff*2), float64(5*time.Minute)))
continue
}
agentID = newID
break
}
// ID persistent speichern - verzeichnis erstellen falls nötig
if err := os.MkdirAll(filepath.Dir(idFile), 0755); err != nil {
log.Printf("WARNUNG: Verzeichnis für Agent-ID konnte nicht erstellt werden: %v", err)
}
if err := os.WriteFile(idFile, []byte(agentID), 0600); err != nil {
log.Printf("WARNUNG: Agent-ID konnte nicht gespeichert werden: %v", err)
}
log.Printf("Agent registriert mit ID: %s", agentID)
// WebSocket-Client starten
wsClient := ws.NewClient(cfg.BackendURL, agentID, cfg.APIKey, cfg.Insecure)
if err := wsClient.Start(); err != nil {
log.Printf("WebSocket-Client starten fehlgeschlagen: %v", err)
}
defer wsClient.Stop()
// Heartbeat-Loop
ticker := time.NewTicker(time.Duration(cfg.IntervalSeconds) * time.Second)
defer ticker.Stop()
// Ersten Heartbeat sofort senden
sendHeartbeat(c, agentID)
backoff = 5 * time.Second
for range ticker.C {
if err := sendHeartbeat(c, agentID); err != nil {
log.Printf("Heartbeat fehlgeschlagen: %v", err)
backoff = time.Duration(math.Min(float64(backoff*2), float64(5*time.Minute)))
} else {
backoff = 5 * time.Second
}
}
}
func sendHeartbeat(c *client.Client, agentID string) error {
log.Println("Sammle Systemdaten...")
hw := collector.CollectHardware()
cpu := collector.CollectCPU()
mem := collector.CollectMemory()
disks := collector.CollectDisks()
net := collector.CollectNetwork()
svcs := collector.CollectServices()
updates := collector.CollectUpdates()
// Proxmox-spezifische Daten (optional)
proxmox := collector.CollectProxmox()
systemData := map[string]interface{}{
"agent_id": agentID,
"hostname": collector.CollectHostname(),
"opnsense_version": collector.CollectOSVersion(), // Linux/Debian Version
"freebsd_version": "", // Nicht verfügbar auf Linux
"uptime_seconds": collector.CollectUptime(),
"hardware": map[string]interface{}{
"manufacturer": hw.Manufacturer,
"model": hw.Model,
"serial": hw.Serial,
"bios_version": hw.BIOSVersion,
},
"cpu": map[string]interface{}{
"model": cpu.Model,
"cores": cpu.Cores,
"threads": cpu.Threads,
"freq_mhz": cpu.FreqMHz,
"usage_percent": cpu.UsagePercent,
},
"memory": map[string]interface{}{
"total_bytes": mem.TotalBytes,
"used_bytes": mem.UsedBytes,
"free_bytes": mem.FreeBytes,
},
"disks": disks,
"network_interfaces": net,
"services": svcs,
"updates": updates,
}
// Proxmox-Daten hinzufügen falls verfügbar
if proxmox != nil {
systemData["proxmox"] = proxmox
}
// Backup-Daten (PVE Jobs + Task-Historie)
if backups := collector.CollectBackups(); backups != nil {
systemData["backups"] = backups
}
req := map[string]interface{}{
"agent_id": agentID,
"agent_version": Version,
"system_data": systemData,
}
if err := c.Heartbeat(req); err != nil {
return err
}
log.Println("Heartbeat gesendet")
return nil
}
// getLocalIP versucht die primaere IP-Adresse zu ermitteln
func getLocalIP() string {
// Versuche ueber die Network Interfaces die erste nicht-loopback IP
ifaces := collector.CollectNetwork()
for _, iface := range ifaces {
if iface.IP != "" && iface.Status == "up" && iface.Name != "lo" {
return iface.IP
}
}
return ""
}
+24
View File
@@ -0,0 +1,24 @@
[Unit]
Description=RMM Agent Linux
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/etc/rmm
ExecStart=/usr/local/bin/rmm-agent-linux -config /etc/rmm/config.yaml
Restart=always
RestartSec=5
# Security settings
NoNewPrivileges=false
PrivateNetwork=false
PrivateTmp=true
ProtectSystem=false
ProtectHome=false
# Allow access to hardware info and system stats
SupplementaryGroups=
[Install]
WantedBy=multi-user.target
+322
View File
@@ -0,0 +1,322 @@
package ws
import (
"context"
"crypto/tls"
"encoding/json"
"fmt"
"log"
"net/url"
"sync"
"time"
"github.com/gorilla/websocket"
)
type Client struct {
backendURL string
agentID string
apiKey string
insecure bool
conn *websocket.Conn
connMux sync.RWMutex
writeMux sync.Mutex
reconnectInterval time.Duration
pingInterval time.Duration
ctx context.Context
cancel context.CancelFunc
messageHandler *Handler
tunnelManager *TunnelManager
}
type Message struct {
Type string `json:"type"`
ID string `json:"id,omitempty"`
Command string `json:"command,omitempty"`
Event string `json:"event,omitempty"`
Status string `json:"status,omitempty"`
Params map[string]interface{} `json:"params,omitempty"`
Data interface{} `json:"data,omitempty"`
Error string `json:"error,omitempty"`
}
func NewClient(backendURL, agentID, apiKey string, insecure bool) *Client {
ctx, cancel := context.WithCancel(context.Background())
client := &Client{
backendURL: backendURL,
agentID: agentID,
apiKey: apiKey,
insecure: insecure,
reconnectInterval: 5 * time.Second,
pingInterval: 30 * time.Second,
ctx: ctx,
cancel: cancel,
}
client.messageHandler = NewHandler(client)
client.tunnelManager = NewTunnelManager(client)
return client
}
func (c *Client) Start() error {
log.Println("WebSocket Client startet...")
go c.reconnectLoop()
return nil
}
func (c *Client) Stop() {
log.Println("WebSocket Client stoppt...")
c.cancel()
c.connMux.Lock()
if c.conn != nil {
c.conn.Close()
c.conn = nil
}
c.connMux.Unlock()
}
func (c *Client) reconnectLoop() {
backoff := c.reconnectInterval
maxBackoff := 5 * time.Minute
for {
select {
case <-c.ctx.Done():
return
default:
}
if err := c.connect(); err != nil {
log.Printf("WebSocket-Verbindung fehlgeschlagen: %v", err)
select {
case <-time.After(backoff):
backoff = time.Duration(float64(backoff) * 1.5)
if backoff > maxBackoff {
backoff = maxBackoff
}
case <-c.ctx.Done():
return
}
continue
}
connStart := time.Now()
log.Println("WebSocket-Verbindung hergestellt")
// Ping/Pong und Message-Handling starten
c.runConnection()
connDuration := time.Since(connStart)
log.Printf("WebSocket-Verbindung getrennt (Dauer: %v)", connDuration)
// Alte Verbindung explizit schliessen
c.connMux.Lock()
if c.conn != nil {
c.conn.Close()
c.conn = nil
}
c.connMux.Unlock()
// Minimum-Delay zwischen Reconnects (verhindert Tight-Loop)
if connDuration < 5*time.Second {
backoff = time.Duration(float64(backoff) * 1.5)
if backoff > maxBackoff {
backoff = maxBackoff
}
log.Printf("Verbindung zu kurz, warte %v vor Reconnect", backoff)
select {
case <-time.After(backoff):
case <-c.ctx.Done():
return
}
} else {
backoff = c.reconnectInterval
}
}
}
func (c *Client) connect() error {
// WebSocket-URL aufbauen
u, err := url.Parse(c.backendURL)
if err != nil {
return fmt.Errorf("Backend-URL parsen: %w", err)
}
if u.Scheme == "https" {
u.Scheme = "wss"
} else {
u.Scheme = "ws"
}
u.Path = "/api/v1/agent/ws"
q := u.Query()
q.Set("agent_id", c.agentID)
q.Set("api_key", c.apiKey)
u.RawQuery = q.Encode()
// Dialer mit TLS-Config
dialer := websocket.Dialer{
HandshakeTimeout: 10 * time.Second,
TLSClientConfig: &tls.Config{
InsecureSkipVerify: c.insecure,
},
}
conn, _, err := dialer.Dial(u.String(), nil)
if err != nil {
return fmt.Errorf("WebSocket-Dial: %w", err)
}
c.connMux.Lock()
c.conn = conn
c.connMux.Unlock()
return nil
}
func (c *Client) runConnection() {
// Ping-Ticker starten
pingTicker := time.NewTicker(c.pingInterval)
defer pingTicker.Stop()
// Pong-Handler setzen
c.connMux.RLock()
conn := c.conn
c.connMux.RUnlock()
if conn == nil {
return
}
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(c.pingInterval * 2))
return nil
})
conn.SetReadDeadline(time.Now().Add(c.pingInterval * 2))
// Goroutine für eingehende Nachrichten
msgChan := make(chan Message, 10)
errChan := make(chan error, 1)
go c.readMessages(conn, msgChan, errChan)
for {
select {
case <-c.ctx.Done():
return
case <-pingTicker.C:
c.writeMux.Lock()
c.connMux.RLock()
conn := c.conn
c.connMux.RUnlock()
if conn != nil {
if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
c.writeMux.Unlock()
log.Printf("Ping senden fehlgeschlagen: %v", err)
return
}
}
c.writeMux.Unlock()
case msg := <-msgChan:
go c.handleMessage(msg)
case err := <-errChan:
log.Printf("WebSocket-Lesefehler: %v", err)
return
}
}
}
func (c *Client) readMessages(conn *websocket.Conn, msgChan chan<- Message, errChan chan<- error) {
defer close(msgChan)
for {
messageType, data, err := conn.ReadMessage()
if err != nil {
errChan <- err
return
}
if messageType == websocket.BinaryMessage {
if err := c.tunnelManager.ProcessBinaryMessage(data); err != nil {
log.Printf("Binary-Message verarbeiten fehlgeschlagen: %v", err)
}
continue
}
// JSON-Message parsen
var msg Message
if err := json.Unmarshal(data, &msg); err != nil {
log.Printf("JSON-Message parsen fehlgeschlagen: %v", err)
continue
}
select {
case msgChan <- msg:
case <-c.ctx.Done():
return
}
}
}
func (c *Client) handleMessage(msg Message) {
switch msg.Type {
case "command":
c.messageHandler.HandleCommand(msg)
case "event":
// Events vom Backend verarbeiten (falls nötig)
log.Printf("Event erhalten: %s", msg.Event)
default:
log.Printf("Unbekannter Nachrichtentyp: %s", msg.Type)
}
}
func (c *Client) SendMessage(msg Message) error {
c.connMux.RLock()
conn := c.conn
c.connMux.RUnlock()
if conn == nil {
return fmt.Errorf("keine WebSocket-Verbindung")
}
c.writeMux.Lock()
err := conn.WriteJSON(msg)
c.writeMux.Unlock()
return err
}
func (c *Client) SendBinaryData(data []byte) error {
c.connMux.RLock()
conn := c.conn
c.connMux.RUnlock()
if conn == nil {
return fmt.Errorf("keine WebSocket-Verbindung")
}
c.writeMux.Lock()
err := conn.WriteMessage(websocket.BinaryMessage, data)
c.writeMux.Unlock()
return err
}
func (c *Client) IsConnected() bool {
c.connMux.RLock()
connected := c.conn != nil
c.connMux.RUnlock()
return connected
}
+261
View File
@@ -0,0 +1,261 @@
package ws
import (
"context"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"fmt"
"log"
"os"
"os/exec"
"strconv"
"time"
)
type Handler struct {
client *Client
}
func NewHandler(client *Client) *Handler {
return &Handler{client: client}
}
func (h *Handler) HandleCommand(msg Message) {
log.Printf("Command erhalten: %s (ID: %s)", msg.Command, msg.ID)
var response Message
response.Type = "response"
response.ID = msg.ID
switch msg.Command {
case "exec":
response = h.handleExec(msg)
case "reboot":
response = h.handleReboot(msg)
case "tunnel_connect":
response = h.handleTunnelConnect(msg)
case "tunnel_disconnect":
h.handleTunnelDisconnect(msg)
return
case "agent_update":
response = h.handleAgentUpdate(msg)
default:
response.Status = "error"
response.Error = fmt.Sprintf("Unbekanntes Command: %s", msg.Command)
}
if err := h.client.SendMessage(response); err != nil {
log.Printf("Response senden fehlgeschlagen: %v", err)
}
}
// handleExec fuehrt einen beliebigen Befehl aus
func (h *Handler) handleExec(msg Message) Message {
response := Message{Type: "response", ID: msg.ID}
command, ok := msg.Params["command"].(string)
if !ok || command == "" {
response.Status = "error"
response.Error = "Parameter 'command' erforderlich"
return response
}
timeoutSec := 30
if t, ok := msg.Params["timeout"]; ok {
if ts, ok := t.(float64); ok {
timeoutSec = int(ts)
} else if ts, ok := t.(string); ok {
if parsed, err := strconv.Atoi(ts); err == nil {
timeoutSec = parsed
}
}
}
output, err := h.runCommand(command, timeoutSec)
if err != nil {
response.Status = "error"
response.Error = err.Error()
response.Data = map[string]interface{}{"output": output}
} else {
response.Status = "ok"
response.Data = map[string]interface{}{"output": output}
}
return response
}
// handleReboot fuehrt einen Reboot durch
func (h *Handler) handleReboot(msg Message) Message {
response := Message{Type: "response", ID: msg.ID}
delaySec := 5
if d, ok := msg.Params["delay"].(float64); ok && d > 0 {
delaySec = int(d)
}
log.Printf("Reboot angefordert (Verzoegerung: %ds)", delaySec)
response.Status = "ok"
response.Data = map[string]interface{}{
"message": fmt.Sprintf("Reboot in %d Sekunden", delaySec),
}
go func() {
time.Sleep(time.Duration(delaySec) * time.Second)
log.Println("Reboot wird ausgefuehrt...")
exec.Command("/sbin/shutdown", "-r", "now").Run()
}()
return response
}
// handleAgentUpdate aktualisiert den Agent
func (h *Handler) handleAgentUpdate(msg Message) Message {
response := Message{Type: "response", ID: msg.ID}
// Binary kommt als Base64 im Data-Feld
data, ok := msg.Data.(map[string]interface{})
if !ok {
response.Status = "error"
response.Error = "Ungueltige Daten"
return response
}
binaryB64, _ := data["binary"].(string)
expectedHash, _ := data["hash"].(string)
newVersion, _ := data["version"].(string)
if binaryB64 == "" || expectedHash == "" {
response.Status = "error"
response.Error = "Binary oder Hash fehlt"
return response
}
log.Printf("Agent-Update empfangen: Version %s, Hash %s", newVersion, expectedHash[:16])
// Base64 dekodieren
binary, err := base64.StdEncoding.DecodeString(binaryB64)
if err != nil {
response.Status = "error"
response.Error = fmt.Sprintf("Base64-Dekodierung fehlgeschlagen: %v", err)
return response
}
// Hash pruefen
hash := sha256.Sum256(binary)
actualHash := hex.EncodeToString(hash[:])
if actualHash != expectedHash {
response.Status = "error"
response.Error = fmt.Sprintf("Hash-Mismatch: erwartet %s, bekommen %s", expectedHash[:16], actualHash[:16])
return response
}
// Aktuelles Binary finden
binaryPath := "/usr/local/bin/rmm-agent-linux"
if _, err := os.Stat(binaryPath); os.IsNotExist(err) {
// Fallback: eigener Pfad
binaryPath, _ = os.Executable()
}
// Backup vom alten Binary
backupPath := binaryPath + ".old"
if err := os.Rename(binaryPath, backupPath); err != nil {
response.Status = "error"
response.Error = fmt.Sprintf("Backup fehlgeschlagen: %v", err)
return response
}
// Neues Binary schreiben
if err := os.WriteFile(binaryPath, binary, 0755); err != nil {
// Rollback
os.Rename(backupPath, binaryPath)
response.Status = "error"
response.Error = fmt.Sprintf("Binary schreiben fehlgeschlagen: %v", err)
return response
}
log.Printf("Neues Binary geschrieben: %s (%d bytes)", binaryPath, len(binary))
response.Status = "ok"
response.Data = map[string]interface{}{
"message": fmt.Sprintf("Update auf %s erfolgreich, Neustart...", newVersion),
"old_version": data["old_version"],
"new_version": newVersion,
}
// Response senden, dann Neustart
go func() {
time.Sleep(2 * time.Second)
log.Printf("Agent-Neustart nach Update auf %s", newVersion)
// Versuche systemd service restart, fallback zu kill + restart
if err := exec.Command("systemctl", "restart", "rmm-agent-linux").Run(); err != nil {
log.Printf("systemctl restart fehlgeschlagen, versuche manuelle Wiederbelebung: %v", err)
// Als letzter Ausweg: sich selbst ersetzen
os.Exit(0)
}
}()
return response
}
// --- Tunnel Handlers ---
func (h *Handler) handleTunnelConnect(msg Message) Message {
response := Message{Type: "response", ID: msg.ID}
sessionID, _ := msg.Params["session_id"].(string)
tunnelID, _ := msg.Params["tunnel_id"].(string)
targetHost, _ := msg.Params["target_host"].(string)
targetPortFloat, _ := msg.Params["target_port"].(float64)
targetPort := int(targetPortFloat)
if sessionID == "" || targetHost == "" || targetPort <= 0 {
response.Status = "error"
response.Error = "session_id, target_host und target_port erforderlich"
return response
}
if err := h.client.tunnelManager.ConnectSession(sessionID, tunnelID, targetHost, targetPort); err != nil {
response.Status = "error"
response.Error = fmt.Sprintf("Session-Connect fehlgeschlagen: %v", err)
return response
}
response.Status = "ok"
response.Data = map[string]interface{}{"session_id": sessionID}
return response
}
func (h *Handler) handleTunnelDisconnect(msg Message) {
sessionID, _ := msg.Params["session_id"].(string)
if sessionID != "" {
h.client.tunnelManager.DisconnectSession(sessionID)
}
}
// SendSessionData schickt Tunnel-Daten als Binary-Message zurueck zum Backend
func (h *Handler) SendSessionData(sessionID string, data []byte) error {
idBytes := []byte(sessionID)
if len(idBytes) > 255 {
return fmt.Errorf("session_id zu lang")
}
message := make([]byte, 1+len(idBytes)+len(data))
message[0] = byte(len(idBytes))
copy(message[1:], idBytes)
copy(message[1+len(idBytes):], data)
return h.client.SendBinaryData(message)
}
// runCommand fuehrt einen Shell-Befehl mit Timeout aus
func (h *Handler) runCommand(command string, timeoutSec int) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second)
defer cancel()
cmd := exec.CommandContext(ctx, "/bin/sh", "-c", command)
output, err := cmd.CombinedOutput()
return string(output), err
}
+178
View File
@@ -0,0 +1,178 @@
package ws
import (
"crypto/rand"
"encoding/hex"
"fmt"
"io"
"log"
"net"
"sync"
"time"
)
// TunnelSession repraesentiert eine einzelne TCP-Verbindung zu einem Ziel
type TunnelSession struct {
ID string
TunnelID string
TargetHost string
TargetPort int
Conn net.Conn
Active bool
}
// TunnelManager verwaltet alle Sessions auf Agent-Seite
type TunnelManager struct {
client *Client
sessions map[string]*TunnelSession
mutex sync.RWMutex
}
func NewTunnelManager(client *Client) *TunnelManager {
return &TunnelManager{
client: client,
sessions: make(map[string]*TunnelSession),
}
}
// ConnectSession oeffnet eine neue TCP-Verbindung fuer eine Session
func (tm *TunnelManager) ConnectSession(sessionID, tunnelID, targetHost string, targetPort int) error {
target := fmt.Sprintf("%s:%d", targetHost, targetPort)
conn, err := net.DialTimeout("tcp", target, 10*time.Second)
if err != nil {
return fmt.Errorf("Verbindung zu %s fehlgeschlagen: %w", target, err)
}
session := &TunnelSession{
ID: sessionID,
TunnelID: tunnelID,
TargetHost: targetHost,
TargetPort: targetPort,
Conn: conn,
Active: true,
}
tm.mutex.Lock()
tm.sessions[sessionID] = session
tm.mutex.Unlock()
// Ziel -> Backend Forwarding starten
go tm.forwardFromTarget(session)
log.Printf("Session %s: Verbindung zu %s hergestellt", sessionID, target)
return nil
}
// DisconnectSession schliesst eine Session
func (tm *TunnelManager) DisconnectSession(sessionID string) {
tm.mutex.Lock()
session, exists := tm.sessions[sessionID]
if !exists {
tm.mutex.Unlock()
return
}
session.Active = false
delete(tm.sessions, sessionID)
tm.mutex.Unlock()
if session.Conn != nil {
session.Conn.Close()
}
log.Printf("Session %s: geschlossen", sessionID)
}
// SendDataToSession schreibt Daten in die Ziel-Verbindung einer Session
func (tm *TunnelManager) SendDataToSession(sessionID string, data []byte) error {
tm.mutex.RLock()
session, exists := tm.sessions[sessionID]
tm.mutex.RUnlock()
if !exists || !session.Active {
return fmt.Errorf("Session %s nicht aktiv", sessionID)
}
_, err := session.Conn.Write(data)
if err != nil {
tm.DisconnectSession(sessionID)
return fmt.Errorf("Session %s: Write fehlgeschlagen: %w", sessionID, err)
}
return nil
}
// forwardFromTarget liest Daten vom Ziel und schickt sie als Binary-Message ans Backend
func (tm *TunnelManager) forwardFromTarget(session *TunnelSession) {
defer func() {
if session.Active {
tm.DisconnectSession(session.ID)
}
log.Printf("Session %s: Forwarding beendet", session.ID)
}()
buffer := make([]byte, 32768)
for session.Active {
session.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
n, err := session.Conn.Read(buffer)
if err != nil {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
continue
}
if err == io.EOF {
log.Printf("Session %s: Ziel-Verbindung geschlossen", session.ID)
} else if session.Active {
log.Printf("Session %s: Read-Fehler: %v", session.ID, err)
}
break
}
if n > 0 {
if err := tm.client.messageHandler.SendSessionData(session.ID, buffer[:n]); err != nil {
log.Printf("Session %s: WebSocket senden fehlgeschlagen: %v", session.ID, err)
break
}
}
}
}
// ProcessBinaryMessage verarbeitet eingehende Binary-Messages vom Backend (Proxy -> Ziel)
func (tm *TunnelManager) ProcessBinaryMessage(data []byte) error {
sessionID, payload, err := ParseBinaryMessage(data)
if err != nil {
return fmt.Errorf("Binary-Message parsen: %w", err)
}
return tm.SendDataToSession(sessionID, payload)
}
// GetActiveSessions gibt die Anzahl aktiver Sessions zurueck
func (tm *TunnelManager) GetActiveSessions() int {
tm.mutex.RLock()
defer tm.mutex.RUnlock()
return len(tm.sessions)
}
func (tm *TunnelManager) generateID() string {
bytes := make([]byte, 8)
rand.Read(bytes)
return hex.EncodeToString(bytes)
}
// ParseBinaryMessage parst das Binary-Format: [id_len:1][id][data]
func ParseBinaryMessage(data []byte) (id string, payload []byte, err error) {
if len(data) < 1 {
return "", nil, fmt.Errorf("Daten zu kurz")
}
idLen := int(data[0])
if len(data) < 1+idLen {
return "", nil, fmt.Errorf("ID zu kurz")
}
id = string(data[1 : 1+idLen])
payload = data[1+idLen:]
return id, payload, nil
}