From ede5cc4f3660f8dd408af1587cc46dd31b3bb03a Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sat, 10 Jan 2026 22:44:10 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=BF=E6=8D=A2=20ssh=20=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E9=87=87=E9=9B=86=E4=B8=BA=20unixgram=20?= =?UTF-8?q?=E9=80=9A=E4=BF=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/agent/main.go | 11 ++ internal/protocol/message.go | 3 +- pkg/agent/service/agent.go | 3 +- pkg/agent/sshmonitor/event.go | 13 ++ pkg/agent/sshmonitor/hook_client.go | 83 ++++++++++ pkg/agent/sshmonitor/hook_manager.go | 153 ++++++++++-------- pkg/agent/sshmonitor/monitor.go | 222 +++++++++------------------ pkg/agent/sshmonitor/parser.go | 64 -------- 8 files changed, 265 insertions(+), 287 deletions(-) create mode 100644 pkg/agent/sshmonitor/event.go create mode 100644 pkg/agent/sshmonitor/hook_client.go delete mode 100644 pkg/agent/sshmonitor/parser.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 2f87f32..cb26e82 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -11,6 +11,7 @@ import ( "github.com/dushixiang/pika/pkg/agent/config" "github.com/dushixiang/pika/pkg/agent/service" + "github.com/dushixiang/pika/pkg/agent/sshmonitor" "github.com/dushixiang/pika/pkg/agent/updater" "github.com/dushixiang/pika/pkg/agent/utils" "github.com/spf13/cobra" @@ -147,6 +148,15 @@ var infoCmd = &cobra.Command{ Run: showInfo, } +var sshLoginHookCmd = &cobra.Command{ + Use: "ssh-login-hook", + Short: "SSH登录监控钩子（PAM 调用）", + Hidden: true, + Run: func(cmd *cobra.Command, args []string) { + _ = sshmonitor.SendEventFromEnv() + }, +} + var ( serverEndpoint string serverAPIKey string @@ -176,6 +186,7 @@ func init() { rootCmd.AddCommand(restartCmd) rootCmd.AddCommand(statusCmd) rootCmd.AddCommand(updateCmd) + rootCmd.AddCommand(sshLoginHookCmd) // 配置命令 configCmd.AddCommand(configInitCmd) diff --git a/internal/protocol/message.go b/internal/protocol/message.go index 53db3fb..2906e7d 100644 --- a/internal/protocol/message.go +++ b/internal/protocol/message.go @@ -684,8 +684,7 @@ type LoginStatistics struct { // SSHLoginConfig SSH登录监控配置 type SSHLoginConfig struct { - Enabled bool `json:"enabled"` // 是否启用监控 - RecordFailed bool `json:"recordFailed"` // 是否记录失败登录 + Enabled bool `json:"enabled"` // 是否启用监控 } // SSHLoginConfigResult SSH登录监控配置应用结果（Agent 反馈） diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index f1055cf..960595a 100644 --- a/pkg/agent/service/agent.go +++ b/pkg/agent/service/agent.go @@ -20,6 +20,7 @@ import ( "github.com/dushixiang/pika/pkg/agent/sshmonitor" "github.com/dushixiang/pika/pkg/agent/tamper" "github.com/dushixiang/pika/pkg/version" + "github.com/gorilla/websocket" "github.com/jpillora/backoff" "github.com/sourcegraph/conc" @@ -874,7 +875,7 @@ func (a *Agent) handleSSHLoginConfig(conn *websocket.Conn, data json.RawMessage) return } - slog.Info("收到SSH登录监控配置", "enabled", sshLoginConfig.Enabled, "recordFailed", sshLoginConfig.RecordFailed) + slog.Info("收到SSH登录监控配置", "enabled", sshLoginConfig.Enabled) // 应用配置 ctx := context.Background() diff --git a/pkg/agent/sshmonitor/event.go b/pkg/agent/sshmonitor/event.go new file mode 100644 index 0000000..d5c1ca7 --- /dev/null +++ b/pkg/agent/sshmonitor/event.go @@ -0,0 +1,13 @@ +package sshmonitor + +// 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"` +} diff --git a/pkg/agent/sshmonitor/hook_client.go b/pkg/agent/sshmonitor/hook_client.go new file mode 100644 index 0000000..f27b1fd --- /dev/null +++ b/pkg/agent/sshmonitor/hook_client.go @@ -0,0 +1,83 @@ +package sshmonitor + +import ( + "encoding/json" + "net" + "os" + "strconv" + "strings" + "time" +) + +// SendEventFromEnv 从 PAM 环境变量构建并发送事件 +func SendEventFromEnv() error { + if os.Getenv("PAM_TYPE") != "open_session" { + return nil + } + event := BuildEventFromEnv() + return SendEvent(event) +} + +// BuildEventFromEnv 从 PAM 环境变量构建事件 +func BuildEventFromEnv() SSHLoginEvent { + now := time.Now().UnixMilli() + + username := os.Getenv("PAM_USER") + if username == "" { + username = "unknown" + } + + ip := os.Getenv("PAM_RHOST") + tty := os.Getenv("PAM_TTY") + if tty == "" { + tty = "unknown" + } + + port := "" + if conn := os.Getenv("SSH_CONNECTION"); conn != "" { + parts := strings.Fields(conn) + if len(parts) >= 2 { + port = parts[1] + } + if ip == "" && len(parts) >= 1 { + ip = parts[0] + } + } + + if ip == "" { + ip = "localhost" + } + + return SSHLoginEvent{ + Username: username, + IP: ip, + Port: port, + Timestamp: now, + Status: "success", + Method: "unknown", + TTY: tty, + SessionID: strconv.Itoa(os.Getpid()), + } +} + +// SendEvent 发送事件到本地 socket +func SendEvent(event SSHLoginEvent) error { + addr := &net.UnixAddr{ + Name: DefaultSocketPath, + Net: "unixgram", + } + + conn, err := net.DialUnix("unixgram", nil, addr) + if err != nil { + return err + } + defer conn.Close() + + data, err := json.Marshal(event) + if err != nil { + return err + } + + _, err = conn.Write(data) + return err +} diff --git a/pkg/agent/sshmonitor/hook_manager.go b/pkg/agent/sshmonitor/hook_manager.go index 9e84c41..48cbfcc 100644 --- a/pkg/agent/sshmonitor/hook_manager.go +++ b/pkg/agent/sshmonitor/hook_manager.go @@ -3,16 +3,18 @@ package sshmonitor import ( "bufio" "fmt" + "io" "log/slog" "os" "os/exec" + "path/filepath" "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" + HookBinaryPath = "/usr/local/bin/pika-agent" + HookCommand = "ssh-login-hook" SSHDConfigFile = "/etc/ssh/sshd_config" UsePAMDirective = "UsePAM yes" ) @@ -25,11 +27,19 @@ func NewHookManager() *HookManager { return &HookManager{} } +func pamConfigLine() string { + return fmt.Sprintf("session optional pam_exec.so %s %s", HookBinaryPath, HookCommand) +} + +func legacyPamConfigLine() string { + return "session optional pam_exec.so /usr/local/bin/pika_ssh_hook.sh" +} + // Install 安装 PAM Hook func (h *HookManager) Install() error { // 检查是否为 root 用户 if os.Geteuid() != 0 { - return fmt.Errorf("安装 PAM Hook 需要 root 权限") + return fmt.Errorf("%w: 安装 PAM Hook 需要 root 权限", os.ErrPermission) } // 检查是否已安装（幂等性） @@ -43,15 +53,15 @@ func (h *HookManager) Install() error { return fmt.Errorf("配置 SSH UsePAM 失败: %w", err) } - // 部署 Hook 脚本 - if err := h.deployScript(); err != nil { - return fmt.Errorf("部署脚本失败: %w", err) + // 安装 Hook 可执行文件（软链） + if err := h.ensureHookBinary(); err != nil { + return fmt.Errorf("安装 Hook 可执行文件失败: %w", err) } // 修改 PAM 配置 if err := h.modifyPAMConfig(true); err != nil { - // 回滚脚本部署 - os.Remove(HookScriptPath) + // 回滚软链 + h.removeHookBinary() return fmt.Errorf("修改 PAM 配置失败: %w", err) } @@ -66,10 +76,8 @@ func (h *HookManager) Uninstall() error { slog.Warn("移除 PAM 配置失败", "error", err) } - // 删除脚本文件 - if err := os.Remove(HookScriptPath); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("删除脚本失败: %w", err) - } + // 删除软链 + h.removeHookBinary() slog.Info("PAM Hook 卸载成功") return nil @@ -87,7 +95,7 @@ func (h *HookManager) isInstalled() bool { scanner := bufio.NewScanner(f) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) - if line == PAMConfigLine { + if line == pamConfigLine() { return true } } @@ -95,58 +103,66 @@ func (h *HookManager) isInstalled() bool { 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 <> "$LOG_FILE" - -exit 0 -` - - // 写入脚本文件 - if err := os.WriteFile(HookScriptPath, []byte(scriptContent), 0755); err != nil { +// ensureHookBinary 确保 Hook 可执行文件存在 +func (h *HookManager) ensureHookBinary() error { + execPath, err := os.Executable() + if err != nil { return err } - slog.Info("PAM Hook 脚本部署成功", "path", HookScriptPath) - return nil + if execPath == HookBinaryPath { + return nil + } + + if err := os.MkdirAll(filepath.Dir(HookBinaryPath), 0755); err != nil { + return err + } + + if info, err := os.Lstat(HookBinaryPath); err == nil { + if info.Mode()&os.ModeSymlink != 0 { + if target, err := os.Readlink(HookBinaryPath); err == nil && target == execPath { + return nil + } + } + } + + _ = os.Remove(HookBinaryPath) + + if err := os.Symlink(execPath, HookBinaryPath); err == nil { + return nil + } + + return copyFile(execPath, HookBinaryPath, 0755) +} + +func (h *HookManager) removeHookBinary() { + info, err := os.Lstat(HookBinaryPath) + if err != nil { + return + } + if info.Mode()&os.ModeSymlink != 0 { + _ = os.Remove(HookBinaryPath) + } +} + +func copyFile(src, dst string, perm os.FileMode) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + + if _, err := io.Copy(out, in); err != nil { + return err + } + + return os.Chmod(dst, perm) } // modifyPAMConfig 修改 PAM 配置 @@ -176,8 +192,13 @@ func (h *HookManager) modifyPAMConfig(add bool) error { for _, line := range lines { trimmed := strings.TrimSpace(line) + // 清理旧版本配置行，避免重复 + if trimmed == legacyPamConfigLine() { + continue + } + // 跳过我们的配置行（移除模式）或检测是否已存在（添加模式） - if trimmed == PAMConfigLine { + if trimmed == pamConfigLine() { found = true if add { newLines = append(newLines, line) // 保留 @@ -191,7 +212,7 @@ func (h *HookManager) modifyPAMConfig(add bool) error { // 如果是添加模式且未找到，则添加到文件末尾 if add && !found { - newLines = append(newLines, PAMConfigLine) + newLines = append(newLines, pamConfigLine()) } // 备份原配置 @@ -280,9 +301,7 @@ func (h *HookManager) ensureUsePAM() error { } // 如果找到了 UsePAM 但未启用,或者没找到,都需要添加配置 - if !usePAMEnabled { - newLines = append(newLines, UsePAMDirective) - } + newLines = append(newLines, UsePAMDirective) // 备份原配置 backupPath := SSHDConfigFile + ".bak" diff --git a/pkg/agent/sshmonitor/monitor.go b/pkg/agent/sshmonitor/monitor.go index 9923d6c..7235f03 100644 --- a/pkg/agent/sshmonitor/monitor.go +++ b/pkg/agent/sshmonitor/monitor.go @@ -2,44 +2,42 @@ package sshmonitor import ( "context" + "encoding/json" + "errors" "fmt" "log/slog" + "net" "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" + DefaultSocketDir = "/run/pika" + DefaultSocketPath = "/run/pika/ssh_login.sock" ) // Monitor SSH登录监控器 type Monitor struct { mu sync.RWMutex enabled bool - logFile string - watcher *fsnotify.Watcher + socketPath string + listener *net.UnixConn 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, + socketPath: DefaultSocketPath, eventCh: make(chan SSHLoginEvent, 100), - parser: NewParser(), hookManager: NewHookManager(), } } @@ -56,7 +54,7 @@ func (m *Monitor) Start(ctx context.Context, config protocol.SSHLoginConfig) err // 如果已启动，先停止 if m.enabled { - m.stopInternal() + _ = m.stopInternal() } if !config.Enabled { @@ -64,20 +62,10 @@ func (m *Monitor) Start(ctx context.Context, config protocol.SSHLoginConfig) err 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) + if err := m.initSocket(ctx); err != nil { + return err } + m.enabled = true // 安装 PAM Hook if err := m.hookManager.Install(); err != nil { @@ -87,22 +75,12 @@ func (m *Monitor) Start(ctx context.Context, config protocol.SSHLoginConfig) err // 不返回错误，继续监控（假设已手动安装） } else { slog.Warn("安装 PAM Hook 失败", "error", err) + _ = m.stopInternal() 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) + slog.Info("SSH登录监控已启动", "socket", m.socketPath) return nil } @@ -125,11 +103,14 @@ func (m *Monitor) stopInternal() error { m.cancel = nil } - // 关闭 watcher - if m.watcher != nil { - m.watcher.Close() - m.watcher = nil - m.watcherOnce = sync.Once{} + // 关闭 socket + if m.listener != nil { + _ = m.listener.Close() + m.listener = nil + } + + if err := os.Remove(m.socketPath); err != nil && !os.IsNotExist(err) { + slog.Warn("移除 socket 失败", "error", err, "path", m.socketPath) } // 卸载 PAM Hook @@ -147,101 +128,69 @@ 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 - } +// initSocket 初始化 socket 监听 +func (m *Monitor) initSocket(ctx context.Context) error { + if err := os.MkdirAll(DefaultSocketDir, 0755); err != nil { + return fmt.Errorf("创建 socket 目录失败: %w", err) + } - // 添加日志文件到监控 - if err = m.watcher.Add(m.logFile); err != nil { - err = fmt.Errorf("添加文件监控失败: %w", err) - return - } + if err := os.Remove(m.socketPath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("清理旧 socket 失败: %w", err) + } - // 创建 context - m.ctx, m.cancel = context.WithCancel(ctx) + addr := &net.UnixAddr{Name: m.socketPath, Net: "unixgram"} + conn, err := net.ListenUnixgram("unixgram", addr) + if err != nil { + return fmt.Errorf("监听 socket 失败: %w", err) + } - // 启动监控循环 - go m.watchLoop() + if err := os.Chmod(m.socketPath, 0600); err != nil { + slog.Warn("设置 socket 权限失败", "error", err, "path", m.socketPath) + } - slog.Info("文件监控器已启动", "file", m.logFile) - }) - return err + m.ctx, m.cancel = context.WithCancel(ctx) + m.listener = conn + + go m.readLoop() + + slog.Info("SSH登录 socket 已启动", "path", m.socketPath) + return nil } -// watchLoop 监控循环 -func (m *Monitor) watchLoop() { +func (m *Monitor) readLoop() { + buf := make([]byte, 4096) for { - select { - case <-m.ctx.Done(): - return - case event, ok := <-m.watcher.Events: - if !ok { + n, _, err := m.listener.ReadFromUnix(buf) + if err != nil { + if errors.Is(err, net.ErrClosed) || errors.Is(err, os.ErrClosed) { 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 { + select { + case <-m.ctx.Done(): return + default: } - slog.Warn("文件监控错误", "error", err) + slog.Warn("读取SSH登录事件失败", "error", err) + continue } - } -} -// handleFileWrite 处理文件写入事件 -func (m *Monitor) handleFileWrite() { - // 打开文件，从上次读取位置继续读 - f, err := os.Open(m.logFile) - if err != nil { - slog.Warn("打开日志文件失败", "error", err) - return - } - defer f.Close() + if n == 0 { + continue + } - // 移动到上次读取位置 - if _, err := f.Seek(m.lastOffset, 0); err != nil { - slog.Warn("移动文件位置失败", "error", err) - return - } + var event SSHLoginEvent + if err := json.Unmarshal(buf[:n], &event); err != nil { + slog.Warn("解析SSH登录事件失败", "error", err) + continue + } - // 解析新增的日志行 - events, newOffset, err := m.parser.ParseFromReader(f, m.lastOffset) - if err != nil { - slog.Warn("解析日志失败", "error", err) - return - } + if event.Timestamp == 0 { + event.Timestamp = time.Now().UnixMilli() + } + if event.Status == "" { + event.Status = "success" + } - // 更新偏移量 - m.lastOffset = newOffset - - // 发送事件 - for _, event := range events { select { case m.eventCh <- event: slog.Info("检测到SSH登录", "user", event.Username, "ip", event.IP, "status", event.Status) @@ -250,36 +199,3 @@ func (m *Monitor) handleFileWrite() { } } } - -// 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 -} diff --git a/pkg/agent/sshmonitor/parser.go b/pkg/agent/sshmonitor/parser.go deleted file mode 100644 index 393ca6e..0000000 --- a/pkg/agent/sshmonitor/parser.go +++ /dev/null @@ -1,64 +0,0 @@ -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 -}