增加 SSH 登录监控

pkg/agent/service/agent.go

@@ -17,6 +17,7 @@ import (
 	"github.com/dushixiang/pika/pkg/agent/collector"
 	"github.com/dushixiang/pika/pkg/agent/config"
 	"github.com/dushixiang/pika/pkg/agent/id"
+	"github.com/dushixiang/pika/pkg/agent/sshmonitor"
 	"github.com/dushixiang/pika/pkg/agent/tamper"
 	"github.com/dushixiang/pika/pkg/version"
 	"github.com/gorilla/websocket"
@@ -70,6 +71,7 @@ type Agent struct {
 	collectorMu      sync.RWMutex
 	collectorManager *collector.Manager
 	tamperProtector  *tamper.Protector
+	sshMonitor       *sshmonitor.Monitor
 }
 
 // New 创建 Agent 实例
@@ -78,6 +80,7 @@ func New(cfg *config.Config) *Agent {
 		cfg:             cfg,
 		idMgr:           id.NewManager(),
 		tamperProtector: tamper.NewProtector(),
+		sshMonitor:      sshmonitor.NewMonitor(),
 	}
 }
 
@@ -237,6 +240,11 @@ func (a *Agent) runOnce(ctx context.Context, onConnected func()) error {
 		a.tamperAlertLoop(ctx, conn, done)
 	})
 
+	// 启动 SSH 登录事件监控
+	wg.Go(func() {
+		a.sshLoginEventLoop(ctx, conn, done)
+	})
+
 	// 等待第一个错误或上下文取消
 	var returnErr error
 	select {
@@ -296,6 +304,8 @@ func (a *Agent) readLoop(conn *websocket.Conn, done chan struct{}) error {
 			go a.handleTamperProtect(msg.Data)
 		case protocol.MessageTypeDDNSConfig:
 			go a.handleDDNSConfig(msg.Data)
+		case protocol.MessageTypeSSHLoginConfig:
+			go a.handleSSHLoginConfig(conn, msg.Data)
 		case protocol.MessageTypeUninstall:
 			go a.handleUninstall()
 		default:
@@ -855,3 +865,89 @@ func (a *Agent) handleUninstall() {
 		a.cancel()
 	}
 }
+
+// handleSSHLoginConfig 处理 SSH 登录监控配置
+func (a *Agent) handleSSHLoginConfig(conn *websocket.Conn, data json.RawMessage) {
+	var config protocol.SSHLoginConfig
+	if err := json.Unmarshal(data, &config); err != nil {
+		slog.Warn("解析SSH登录监控配置失败", "error", err)
+		// 发送失败结果
+		a.sendSSHLoginConfigResult(conn, false, false, "解析配置失败", err.Error())
+		return
+	}
+
+	slog.Info("收到SSH登录监控配置", "enabled", config.Enabled, "recordFailed", config.RecordFailed)
+
+	// 应用配置
+	ctx := context.Background()
+	if err := a.sshMonitor.Start(ctx, config); err != nil {
+		slog.Warn("应用SSH登录监控配置失败", "error", err)
+		// 发送失败结果，包含详细错误信息
+		a.sendSSHLoginConfigResult(conn, false, config.Enabled, "应用配置失败", err.Error())
+		return
+	}
+
+	// 发送成功结果
+	message := "配置已成功应用"
+	if config.Enabled {
+		message = "SSH登录监控已启用"
+	} else {
+		message = "SSH登录监控已禁用"
+	}
+	a.sendSSHLoginConfigResult(conn, true, config.Enabled, message, "")
+}
+
+// sendSSHLoginConfigResult 发送 SSH 登录监控配置应用结果
+func (a *Agent) sendSSHLoginConfigResult(conn *websocket.Conn, success bool, enabled bool, message string, errorMsg string) {
+	result := protocol.SSHLoginConfigResult{
+		Success: success,
+		Enabled: enabled,
+		Message: message,
+		Error:   errorMsg,
+	}
+
+	msg := protocol.OutboundMessage{
+		Type: protocol.MessageTypeSSHLoginConfigResult,
+		Data: result,
+	}
+
+	if err := conn.WriteJSON(msg); err != nil {
+		slog.Warn("发送SSH登录监控配置应用结果失败", "error", err)
+	}
+}
+
+// sshLoginEventLoop SSH登录事件监控循环
+func (a *Agent) sshLoginEventLoop(ctx context.Context, conn *safeConn, done chan struct{}) {
+	eventCh := a.sshMonitor.GetEvents()
+
+	for {
+		select {
+		case <-done:
+			return
+		case <-ctx.Done():
+			return
+		case event := <-eventCh:
+			// 转换为 protocol 结构
+			eventData := protocol.SSHLoginEvent{
+				Username:  event.Username,
+				IP:        event.IP,
+				Port:      event.Port,
+				Timestamp: event.Timestamp,
+				Status:    event.Status,
+				Method:    event.Method,
+				TTY:       event.TTY,
+				SessionID: event.SessionID,
+			}
+
+			// 上报到服务端
+			if err := conn.WriteJSON(protocol.OutboundMessage{
+				Type: protocol.MessageTypeSSHLoginEvent,
+				Data: eventData,
+			}); err != nil {
+				slog.Warn("发送SSH登录事件失败", "error", err)
+			} else {
+				slog.Info("已上报SSH登录事件", "user", event.Username, "ip", event.IP, "status", event.Status)
+			}
+		}
+	}
+}

pkg/agent/service/service.go

@@ -12,6 +12,7 @@ import (
 	"github.com/dushixiang/pika/pkg/agent"
 	"github.com/dushixiang/pika/pkg/agent/config"
 	"github.com/dushixiang/pika/pkg/agent/id"
+	"github.com/dushixiang/pika/pkg/agent/sshmonitor"
 	"github.com/dushixiang/pika/pkg/agent/sysutil"
 	"github.com/dushixiang/pika/pkg/agent/updater"
 	"github.com/kardianos/service"
@@ -324,6 +325,12 @@ func UninstallAgent(cfgPath string) error {
 		return fmt.Errorf("卸载服务失败: %w", err)
 	}
 
+	// 清理 SSH 监控配置
+	monitor := sshmonitor.NewMonitor()
+	if err := monitor.Stop(); err != nil {
+		slog.Warn("清理SSH监控配置失败", "error", err)
+	}
+
 	// 删除配置文件
 	if err := os.Remove(cfgPath); err != nil {
 		slog.Warn("删除配置文件失败", "error", err)

pkg/agent/sshmonitor/hook_manager.go

@@ -0,0 +1,220 @@
+package sshmonitor
+
+import (
+	"bufio"
+	"fmt"
+	"log/slog"
+	"os"
+	"strings"
+)
+
+const (
+	PAMConfigFile  = "/etc/pam.d/sshd"
+	PAMConfigLine  = "session optional pam_exec.so /usr/local/bin/pika_ssh_hook.sh"
+	HookScriptPath = "/usr/local/bin/pika_ssh_hook.sh"
+)
+
+// HookManager PAM Hook 管理器
+type HookManager struct{}
+
+// NewHookManager 创建管理器
+func NewHookManager() *HookManager {
+	return &HookManager{}
+}
+
+// Install 安装 PAM Hook
+func (h *HookManager) Install() error {
+	// 检查是否为 root 用户
+	if os.Geteuid() != 0 {
+		return fmt.Errorf("安装 PAM Hook 需要 root 权限")
+	}
+
+	// 检查是否已安装（幂等性）
+	if h.isInstalled() {
+		slog.Info("PAM Hook 已安装，跳过")
+		return nil
+	}
+
+	// 部署 Hook 脚本
+	if err := h.deployScript(); err != nil {
+		return fmt.Errorf("部署脚本失败: %w", err)
+	}
+
+	// 修改 PAM 配置
+	if err := h.modifyPAMConfig(true); err != nil {
+		// 回滚脚本部署
+		os.Remove(HookScriptPath)
+		return fmt.Errorf("修改 PAM 配置失败: %w", err)
+	}
+
+	slog.Info("PAM Hook 安装成功")
+	return nil
+}
+
+// Uninstall 卸载 PAM Hook
+func (h *HookManager) Uninstall() error {
+	// 移除 PAM 配置
+	if err := h.modifyPAMConfig(false); err != nil {
+		slog.Warn("移除 PAM 配置失败", "error", err)
+	}
+
+	// 删除脚本文件
+	if err := os.Remove(HookScriptPath); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("删除脚本失败: %w", err)
+	}
+
+	slog.Info("PAM Hook 卸载成功")
+	return nil
+}
+
+// isInstalled 检查是否已安装
+func (h *HookManager) isInstalled() bool {
+	// 检查 PAM 配置文件是否包含我们的配置行
+	f, err := os.Open(PAMConfigFile)
+	if err != nil {
+		return false
+	}
+	defer f.Close()
+
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == PAMConfigLine {
+			return true
+		}
+	}
+
+	return false
+}
+
+// deployScript 部署 Hook 脚本
+func (h *HookManager) deployScript() error {
+	// 脚本内容（嵌入式）
+	scriptContent := `#!/bin/bash
+# Pika SSH Login Hook
+# 由 Pika Agent 自动安装
+
+LOG_FILE="/var/log/pika/ssh_login.log"
+LOG_DIR="/var/log/pika"
+
+# 确保日志目录存在
+mkdir -p "$LOG_DIR"
+
+# 获取登录信息
+TIMESTAMP=$(date +%s)000  # 毫秒时间戳
+USERNAME="$PAM_USER"
+IP="${PAM_RHOST:-localhost}"
+TTY="${PAM_TTY:-unknown}"
+SESSION_ID="$$"
+
+# 判断登录状态（通过 PAM_TYPE）
+if [ "$PAM_TYPE" = "open_session" ]; then
+    STATUS="success"
+else
+    STATUS="unknown"
+fi
+
+# 尝试获取端口号（从 SSH 连接信息）
+PORT=$(echo "$SSH_CONNECTION" | awk '{print $2}')
+
+# 尝试获取认证方式（简化处理）
+METHOD="password"
+
+# 构建 JSON 日志
+JSON_LOG=$(cat <<EOF
+{"timestamp":$TIMESTAMP,"username":"$USERNAME","ip":"$IP","port":"$PORT","status":"$STATUS","method":"$METHOD","tty":"$TTY","sessionId":"$SESSION_ID"}
+EOF
+)
+
+# 追加到日志文件
+echo "$JSON_LOG" >> "$LOG_FILE"
+
+exit 0
+`
+
+	// 写入脚本文件
+	if err := os.WriteFile(HookScriptPath, []byte(scriptContent), 0755); err != nil {
+		return err
+	}
+
+	slog.Info("PAM Hook 脚本部署成功", "path", HookScriptPath)
+	return nil
+}
+
+// modifyPAMConfig 修改 PAM 配置
+// add: true=添加配置, false=移除配置
+func (h *HookManager) modifyPAMConfig(add bool) error {
+	// 读取现有配置
+	f, err := os.Open(PAMConfigFile)
+	if err != nil {
+		return err
+	}
+
+	var lines []string
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		lines = append(lines, scanner.Text())
+	}
+	f.Close()
+
+	if err := scanner.Err(); err != nil {
+		return err
+	}
+
+	// 修改配置
+	var newLines []string
+	found := false
+
+	for _, line := range lines {
+		trimmed := strings.TrimSpace(line)
+
+		// 跳过我们的配置行（移除模式）或检测是否已存在（添加模式）
+		if trimmed == PAMConfigLine {
+			found = true
+			if add {
+				newLines = append(newLines, line) // 保留
+			}
+			// 移除模式：不添加到 newLines
+			continue
+		}
+
+		newLines = append(newLines, line)
+	}
+
+	// 如果是添加模式且未找到，则添加到文件末尾
+	if add && !found {
+		newLines = append(newLines, PAMConfigLine)
+	}
+
+	// 备份原配置
+	backupPath := PAMConfigFile + ".bak"
+	if err := os.Rename(PAMConfigFile, backupPath); err != nil {
+		return err
+	}
+
+	// 写入新配置
+	outF, err := os.Create(PAMConfigFile)
+	if err != nil {
+		// 恢复备份
+		os.Rename(backupPath, PAMConfigFile)
+		return err
+	}
+	defer outF.Close()
+
+	writer := bufio.NewWriter(outF)
+	for _, line := range newLines {
+		writer.WriteString(line + "\n")
+	}
+	if err := writer.Flush(); err != nil {
+		// 恢复备份
+		outF.Close()
+		os.Rename(backupPath, PAMConfigFile)
+		return err
+	}
+
+	// 设置正确的权限
+	os.Chmod(PAMConfigFile, 0644)
+
+	slog.Info("PAM 配置修改成功", "add", add)
+	return nil
+}

pkg/agent/sshmonitor/monitor.go

@@ -0,0 +1,285 @@
+package sshmonitor
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/dushixiang/pika/internal/protocol"
+	"github.com/fsnotify/fsnotify"
+)
+
+const (
+	DefaultLogFile = "/var/log/pika/ssh_login.log"
+	DefaultLogDir  = "/var/log/pika"
+)
+
+// Monitor SSH登录监控器
+type Monitor struct {
+	mu          sync.RWMutex
+	enabled     bool
+	logFile     string
+	watcher     *fsnotify.Watcher
+	ctx         context.Context
+	cancel      context.CancelFunc
+	eventCh     chan SSHLoginEvent
+	parser      *Parser
+	hookManager *HookManager
+	watcherOnce sync.Once
+	lastOffset  int64 // 记录上次读取的位置，避免重复读取
+}
+
+// NewMonitor 创建监控器
+func NewMonitor() *Monitor {
+	return &Monitor{
+		enabled:     false,
+		logFile:     DefaultLogFile,
+		eventCh:     make(chan SSHLoginEvent, 100),
+		parser:      NewParser(),
+		hookManager: NewHookManager(),
+	}
+}
+
+// Start 启动监控（根据配置启用/禁用）
+func (m *Monitor) Start(ctx context.Context, config protocol.SSHLoginConfig) error {
+	// 检查操作系统
+	if runtime.GOOS != "linux" {
+		return fmt.Errorf("SSH登录监控仅支持 Linux 系统")
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// 如果已启动，先停止
+	if m.enabled {
+		m.stopInternal()
+	}
+
+	if !config.Enabled {
+		slog.Info("SSH登录监控已禁用")
+		return nil
+	}
+
+	// 确保日志目录存在
+	if err := os.MkdirAll(DefaultLogDir, 0755); err != nil {
+		return fmt.Errorf("创建日志目录失败: %w", err)
+	}
+
+	// 确保日志文件存在
+	if _, err := os.Stat(m.logFile); os.IsNotExist(err) {
+		f, err := os.Create(m.logFile)
+		if err != nil {
+			return fmt.Errorf("创建日志文件失败: %w", err)
+		}
+		f.Close()
+		os.Chmod(m.logFile, 0644)
+	}
+
+	// 安装 PAM Hook
+	if err := m.hookManager.Install(); err != nil {
+		if os.IsPermission(err) {
+			slog.Warn("权限不足，无法自动安装 PAM Hook", "error", err)
+			slog.Info("如需启用 SSH 登录监控，请手动执行安装")
+			// 不返回错误，继续监控（假设已手动安装）
+		} else {
+			slog.Warn("安装 PAM Hook 失败", "error", err)
+			return err
+		}
+	}
+
+	// 初始化 watcher
+	if err := m.initWatcher(ctx); err != nil {
+		return err
+	}
+
+	// 读取现有日志（避免丢失 Agent 重启期间的登录）
+	if err := m.readExistingLogs(); err != nil {
+		slog.Warn("读取现有日志失败", "error", err)
+	}
+
+	m.enabled = true
+	slog.Info("SSH登录监控已启动", "logFile", m.logFile)
+	return nil
+}
+
+// Stop 停止监控
+func (m *Monitor) Stop() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.stopInternal()
+}
+
+// stopInternal 内部停止方法（不加锁）
+func (m *Monitor) stopInternal() error {
+	if !m.enabled {
+		return nil
+	}
+
+	// 取消 context
+	if m.cancel != nil {
+		m.cancel()
+		m.cancel = nil
+	}
+
+	// 关闭 watcher
+	if m.watcher != nil {
+		m.watcher.Close()
+		m.watcher = nil
+		m.watcherOnce = sync.Once{}
+	}
+
+	// 卸载 PAM Hook
+	if err := m.hookManager.Uninstall(); err != nil {
+		slog.Warn("卸载 PAM Hook 失败", "error", err)
+	}
+
+	m.enabled = false
+	slog.Info("SSH登录监控已停止")
+	return nil
+}
+
+// GetEvents 获取事件通道
+func (m *Monitor) GetEvents() <-chan SSHLoginEvent {
+	return m.eventCh
+}
+
+// initWatcher 初始化文件监控器
+func (m *Monitor) initWatcher(ctx context.Context) error {
+	var err error
+	m.watcherOnce.Do(func() {
+		m.watcher, err = fsnotify.NewWatcher()
+		if err != nil {
+			err = fmt.Errorf("创建文件监控器失败: %w", err)
+			return
+		}
+
+		// 添加日志文件到监控
+		if err = m.watcher.Add(m.logFile); err != nil {
+			err = fmt.Errorf("添加文件监控失败: %w", err)
+			return
+		}
+
+		// 创建 context
+		m.ctx, m.cancel = context.WithCancel(ctx)
+
+		// 启动监控循环
+		go m.watchLoop()
+
+		slog.Info("文件监控器已启动", "file", m.logFile)
+	})
+	return err
+}
+
+// watchLoop 监控循环
+func (m *Monitor) watchLoop() {
+	for {
+		select {
+		case <-m.ctx.Done():
+			return
+		case event, ok := <-m.watcher.Events:
+			if !ok {
+				return
+			}
+
+			// 检测文件被删除或重命名（logrotate）
+			if event.Op&fsnotify.Remove == fsnotify.Remove ||
+				event.Op&fsnotify.Rename == fsnotify.Rename {
+				slog.Warn("日志文件被删除或轮转，尝试重新监控")
+
+				// 等待新文件创建
+				time.Sleep(1 * time.Second)
+
+				// 重新添加监控
+				if err := m.watcher.Add(m.logFile); err != nil {
+					slog.Error("重新添加文件监控失败", "error", err)
+				} else {
+					m.lastOffset = 0 // 重置偏移量
+				}
+			}
+
+			// 只处理写入事件
+			if event.Op&fsnotify.Write == fsnotify.Write {
+				m.handleFileWrite()
+			}
+		case err, ok := <-m.watcher.Errors:
+			if !ok {
+				return
+			}
+			slog.Warn("文件监控错误", "error", err)
+		}
+	}
+}
+
+// handleFileWrite 处理文件写入事件
+func (m *Monitor) handleFileWrite() {
+	// 打开文件，从上次读取位置继续读
+	f, err := os.Open(m.logFile)
+	if err != nil {
+		slog.Warn("打开日志文件失败", "error", err)
+		return
+	}
+	defer f.Close()
+
+	// 移动到上次读取位置
+	if _, err := f.Seek(m.lastOffset, 0); err != nil {
+		slog.Warn("移动文件位置失败", "error", err)
+		return
+	}
+
+	// 解析新增的日志行
+	events, newOffset, err := m.parser.ParseFromReader(f, m.lastOffset)
+	if err != nil {
+		slog.Warn("解析日志失败", "error", err)
+		return
+	}
+
+	// 更新偏移量
+	m.lastOffset = newOffset
+
+	// 发送事件
+	for _, event := range events {
+		select {
+		case m.eventCh <- event:
+			slog.Info("检测到SSH登录", "user", event.Username, "ip", event.IP, "status", event.Status)
+		default:
+			slog.Warn("事件队列已满，丢弃事件")
+		}
+	}
+}
+
+// readExistingLogs 读取现有日志（Agent重启时）
+func (m *Monitor) readExistingLogs() error {
+	f, err := os.Open(m.logFile)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	// 从文件开头读取所有日志
+	events, newOffset, err := m.parser.ParseFromReader(f, 0)
+	if err != nil {
+		return err
+	}
+
+	m.lastOffset = newOffset
+
+	// 只上报最近的登录（例如最近100条），避免重复上报历史数据
+	startIdx := len(events) - 100
+	if startIdx < 0 {
+		startIdx = 0
+	}
+
+	for i := startIdx; i < len(events); i++ {
+		select {
+		case m.eventCh <- events[i]:
+		default:
+			break
+		}
+	}
+
+	return nil
+}

pkg/agent/sshmonitor/parser.go

@@ -0,0 +1,64 @@
+package sshmonitor
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+)
+
+// SSHLoginEvent SSH登录事件（内部结构）
+type SSHLoginEvent struct {
+	Username  string `json:"username"`
+	IP        string `json:"ip"`
+	Port      string `json:"port,omitempty"`
+	Timestamp int64  `json:"timestamp"`
+	Status    string `json:"status"`
+	Method    string `json:"method,omitempty"`
+	TTY       string `json:"tty,omitempty"`
+	SessionID string `json:"sessionId,omitempty"`
+}
+
+// Parser 日志解析器
+type Parser struct{}
+
+// NewParser 创建解析器
+func NewParser() *Parser {
+	return &Parser{}
+}
+
+// ParseFromReader 从 Reader 解析日志
+// 返回：事件列表、新的文件偏移量、错误
+func (p *Parser) ParseFromReader(r io.Reader, startOffset int64) ([]SSHLoginEvent, int64, error) {
+	var events []SSHLoginEvent
+	scanner := bufio.NewScanner(r)
+
+	currentOffset := startOffset
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		lineSize := int64(len(line) + 1) // +1 for newline
+
+		// 跳过空行
+		if line == "" {
+			currentOffset += lineSize
+			continue
+		}
+
+		// 解析 JSON
+		var event SSHLoginEvent
+		if err := json.Unmarshal([]byte(line), &event); err != nil {
+			// 跳过格式错误的行
+			currentOffset += lineSize
+			continue
+		}
+
+		events = append(events, event)
+		currentOffset += lineSize
+	}
+
+	if err := scanner.Err(); err != nil {
+		return events, currentOffset, err
+	}
+
+	return events, currentOffset, nil
+}