okooooooo/pika
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

替换 ssh 登录事件采集为 unixgram 通信

Browse Source
This commit is contained in:
dushixiang committed 2026-01-10 22:44:10 +08:00
1 parent c064d86c68
commit ede5cc4f36
8 files changed
+265 -287
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cmd/agent/main.go
+11
View File
@@ -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)
internal/protocol/message.go
+1 -2
View File
@@ -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 反馈）
pkg/agent/service/agent.go
+2 -1
View File
@@ -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()
pkg/agent/sshmonitor/event.go
+13
View File
@@ -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"`
}
pkg/agent/sshmonitor/hook_client.go
+83
View File
@@ -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
}
pkg/agent/sshmonitor/hook_manager.go
+86 -67
View File
@@ -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 <<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 {
// 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"
pkg/agent/sshmonitor/monitor.go
+69 -153
View File
@@ -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
}
pkg/agent/sshmonitor/parser.go
-64
View File
@@ -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
}