From 9d14a8cba5af0e1411c0d5e0b2b7ac2f49b6f44e Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sat, 3 Jan 2026 11:05:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20SSH=20=E7=99=BB=E5=BD=95?= =?UTF-8?q?=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 2 + internal/app.go | 16 ++ internal/handler/agent_handler.go | 53 +++-- internal/handler/ssh_login_handler.go | 201 ++++++++++++++++ internal/models/ssh_login.go | 60 +++++ internal/protocol/message.go | 32 +++ internal/repo/ssh_login_repo.go | 175 ++++++++++++++ internal/service/ddns_service.go | 8 +- internal/service/ssh_login_service.go | 270 +++++++++++++++++++++ internal/service/tamper_service.go | 4 +- internal/wire.go | 10 +- internal/wire_gen.go | 16 +- pkg/agent/service/agent.go | 96 ++++++++ pkg/agent/service/service.go | 7 + pkg/agent/sshmonitor/hook_manager.go | 220 +++++++++++++++++ pkg/agent/sshmonitor/monitor.go | 285 +++++++++++++++++++++++ pkg/agent/sshmonitor/parser.go | 64 +++++ web/src/api/agent.ts | 52 ++++- web/src/pages/Agents/AgentDetail.tsx | 22 +- web/src/pages/Agents/SSHLoginConfig.tsx | 121 ++++++++++ web/src/pages/Agents/SSHLoginEvents.tsx | 200 ++++++++++++++++ web/src/pages/Agents/SSHLoginMonitor.tsx | 47 ++++ web/src/types/index.ts | 36 +++ 23 files changed, 1959 insertions(+), 38 deletions(-) create mode 100644 internal/handler/ssh_login_handler.go create mode 100644 internal/models/ssh_login.go create mode 100644 internal/repo/ssh_login_repo.go create mode 100644 internal/service/ssh_login_service.go create mode 100644 pkg/agent/sshmonitor/hook_manager.go create mode 100644 pkg/agent/sshmonitor/monitor.go create mode 100644 pkg/agent/sshmonitor/parser.go create mode 100644 web/src/pages/Agents/SSHLoginConfig.tsx create mode 100644 web/src/pages/Agents/SSHLoginEvents.tsx create mode 100644 web/src/pages/Agents/SSHLoginMonitor.tsx diff --git a/.gitignore b/.gitignore index a0dba26..b814920 100644 --- a/.gitignore +++ b/.gitignore @@ -40,3 +40,5 @@ go.work.sum config.yaml agent.yaml playground +web/node_modules +web/dist diff --git a/internal/app.go b/internal/app.go index 5c1a870..08b7f38 100644 --- a/internal/app.go +++ b/internal/app.go @@ -3,6 +3,7 @@ package internal import ( "bytes" "context" + "fmt" "log" "net/http" "strings" @@ -108,6 +109,13 @@ func setupApi(app *orz.App, components *AppComponents) { e.Use(middleware.Recover()) e.Use(ErrorHandler(logger)) + e.Use(middleware.RecoverWithConfig(middleware.RecoverConfig{ + LogErrorFunc: func(c echo.Context, err error, stack []byte) error { + sugar := logger.Named("[PANIC RECOVER]").Sugar() + sugar.Error(fmt.Sprintf("%v %s\n", err, stack)) + return err + }, + })) indexTemplate, err := template.New("index").Parse(web.IndexHtml()) if err != nil { @@ -249,6 +257,12 @@ func setupApi(app *orz.App, components *AppComponents) { adminApi.GET("/agents/:id/tamper/events", components.TamperHandler.GetTamperEvents) adminApi.GET("/agents/:id/tamper/alerts", components.TamperHandler.GetTamperAlerts) + // SSH 登录监控管理（管理员功能） + adminApi.GET("/agents/:id/ssh-login/config", components.SSHLoginHandler.GetConfig) + adminApi.POST("/agents/:id/ssh-login/config", components.SSHLoginHandler.UpdateConfig) + adminApi.GET("/agents/:id/ssh-login/events", components.SSHLoginHandler.ListEvents) + adminApi.DELETE("/agents/:id/ssh-login/events", components.SSHLoginHandler.DeleteEvents) + // 通用属性管理 adminApi.GET("/properties/:id", components.PropertyHandler.GetProperty) adminApi.PUT("/properties/:id", components.PropertyHandler.SetProperty) @@ -306,6 +320,8 @@ func autoMigrate(database *gorm.DB) error { &models.TamperAlert{}, // 防篡改告警 &models.DDNSConfig{}, // DDNS 配置 &models.DDNSRecord{}, // DDNS 记录 + &models.SSHLoginConfig{}, // SSH 登录监控配置 + &models.SSHLoginEvent{}, // SSH 登录事件 ) } diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index 75ad790..cb2d97f 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -24,27 +24,30 @@ import ( ) type AgentHandler struct { - logger *zap.Logger - agentService *service.AgentService - metricService *service.MetricService - monitorSvc *service.MonitorService - tamperService *service.TamperService - ddnsService *service.DDNSService - wsManager *ws.Manager - upgrader websocket.Upgrader + logger *zap.Logger + agentService *service.AgentService + metricService *service.MetricService + monitorSvc *service.MonitorService + tamperService *service.TamperService + ddnsService *service.DDNSService + sshLoginService *service.SSHLoginService + wsManager *ws.Manager + upgrader websocket.Upgrader } func NewAgentHandler(logger *zap.Logger, agentService *service.AgentService, metricService *service.MetricService, - monitorService *service.MonitorService, tamperService *service.TamperService, ddnsService *service.DDNSService, wsManager *ws.Manager) *AgentHandler { + monitorService *service.MonitorService, tamperService *service.TamperService, ddnsService *service.DDNSService, + sshLoginService *service.SSHLoginService, wsManager *ws.Manager) *AgentHandler { h := &AgentHandler{ - logger: logger, - agentService: agentService, - metricService: metricService, - monitorSvc: monitorService, - tamperService: tamperService, - ddnsService: ddnsService, - wsManager: wsManager, + logger: logger, + agentService: agentService, + metricService: metricService, + monitorSvc: monitorService, + tamperService: tamperService, + ddnsService: ddnsService, + sshLoginService: sshLoginService, + wsManager: wsManager, } // 初始化upgrader，需要在创建handler之后因为需要引用h.checkOrigin @@ -199,6 +202,24 @@ func (h *AgentHandler) handleWebSocketMessage(ctx context.Context, agentID strin } return h.ddnsService.HandleIPReport(ctx, agentID, &ipReport) + case protocol.MessageTypeSSHLoginEvent: + // SSH 登录事件 + var eventData protocol.SSHLoginEvent + if err := json.Unmarshal(data, &eventData); err != nil { + h.logger.Error("failed to unmarshal ssh login event", zap.Error(err)) + return err + } + return h.sshLoginService.HandleEvent(agentID, eventData) + + case protocol.MessageTypeSSHLoginConfigResult: + // SSH 登录监控配置应用结果 + var resultData protocol.SSHLoginConfigResult + if err := json.Unmarshal(data, &resultData); err != nil { + h.logger.Error("failed to unmarshal ssh login config result", zap.Error(err)) + return err + } + return h.sshLoginService.HandleConfigResult(agentID, resultData) + case protocol.MessageTypeTamperProtect: // 防篡改配置响应 var protectResp protocol.TamperProtectResponse diff --git a/internal/handler/ssh_login_handler.go b/internal/handler/ssh_login_handler.go new file mode 100644 index 0000000..4095955 --- /dev/null +++ b/internal/handler/ssh_login_handler.go @@ -0,0 +1,201 @@ +package handler + +import ( + "net/http" + "strconv" + + "github.com/dushixiang/pika/internal/service" + "github.com/labstack/echo/v4" + "go.uber.org/zap" +) + +// SSHLoginHandler SSH登录处理器 +type SSHLoginHandler struct { + logger *zap.Logger + service *service.SSHLoginService +} + +// NewSSHLoginHandler 创建处理器 +func NewSSHLoginHandler(logger *zap.Logger, service *service.SSHLoginService) *SSHLoginHandler { + return &SSHLoginHandler{ + logger: logger, + service: service, + } +} + +// GetConfig 获取SSH登录监控配置 +// GET /api/agents/:id/ssh-login/config +func (h *SSHLoginHandler) GetConfig(c echo.Context) error { + agentID := c.Param("id") + if agentID == "" { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "探针ID不能为空", + }) + } + + config, err := h.service.GetConfig(agentID) + if err != nil { + h.logger.Error("获取SSH登录监控配置失败", zap.Error(err)) + return c.JSON(http.StatusInternalServerError, map[string]string{ + "error": "获取配置失败", + }) + } + + if config == nil { + // 返回默认配置 + return c.JSON(http.StatusOK, map[string]interface{}{ + "enabled": false, + "recordFailed": true, + }) + } + + return c.JSON(http.StatusOK, config) +} + +// UpdateConfig 更新SSH登录监控配置 +// POST /api/agents/:id/ssh-login/config +func (h *SSHLoginHandler) UpdateConfig(c echo.Context) error { + agentID := c.Param("id") + if agentID == "" { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "探针ID不能为空", + }) + } + + var req struct { + Enabled bool `json:"enabled"` + RecordFailed bool `json:"recordFailed"` + } + + if err := c.Bind(&req); err != nil { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "请求参数错误", + }) + } + + config, configSent, err := h.service.UpdateConfig(c.Request().Context(), agentID, req.Enabled, req.RecordFailed) + if err != nil { + h.logger.Error("更新SSH登录监控配置失败", zap.Error(err)) + return c.JSON(http.StatusInternalServerError, map[string]string{ + "error": "更新配置失败", + }) + } + + // 根据下发状态返回不同的消息 + message := "配置已保存" + if configSent { + message = "配置已保存并成功下发到探针" + } else { + message = "配置已保存，将在探针下次连接时生效" + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "message": message, + "config": config, + "configSent": configSent, // 告知前端配置是否成功下发 + }) +} + +// ListEvents 查询SSH登录事件 +// GET /api/agents/:id/ssh-login/events +func (h *SSHLoginHandler) ListEvents(c echo.Context) error { + agentID := c.Param("id") + if agentID == "" { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "探针ID不能为空", + }) + } + + // 解析分页参数 + page, _ := strconv.Atoi(c.QueryParam("page")) + if page < 1 { + page = 1 + } + + pageSize, _ := strconv.Atoi(c.QueryParam("pageSize")) + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + // 解析过滤参数 + username := c.QueryParam("username") + ip := c.QueryParam("ip") + status := c.QueryParam("status") + + startTime, _ := strconv.ParseInt(c.QueryParam("startTime"), 10, 64) + endTime, _ := strconv.ParseInt(c.QueryParam("endTime"), 10, 64) + + // 查询事件 + var events interface{} + var total int64 + var err error + + if username != "" || ip != "" || status != "" || startTime > 0 || endTime > 0 { + events, total, err = h.service.ListEventsByFilter(agentID, username, ip, status, startTime, endTime, page, pageSize) + } else { + events, total, err = h.service.ListEvents(agentID, page, pageSize) + } + + if err != nil { + h.logger.Error("查询SSH登录事件失败", zap.Error(err)) + return c.JSON(http.StatusInternalServerError, map[string]string{ + "error": "查询失败", + }) + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "items": events, + "total": total, + "page": page, + "pageSize": pageSize, + }) +} + +// GetEvent 获取单个SSH登录事件 +// GET /api/ssh-login/events/:id +func (h *SSHLoginHandler) GetEvent(c echo.Context) error { + eventID := c.Param("id") + if eventID == "" { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "事件ID不能为空", + }) + } + + event, err := h.service.GetEventByID(eventID) + if err != nil { + h.logger.Error("获取SSH登录事件失败", zap.Error(err)) + return c.JSON(http.StatusInternalServerError, map[string]string{ + "error": "获取事件失败", + }) + } + + if event == nil { + return c.JSON(http.StatusNotFound, map[string]string{ + "error": "事件不存在", + }) + } + + return c.JSON(http.StatusOK, event) +} + +// DeleteEvents 删除探针的所有SSH登录事件 +// DELETE /api/agents/:id/ssh-login/events +func (h *SSHLoginHandler) DeleteEvents(c echo.Context) error { + agentID := c.Param("id") + if agentID == "" { + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "探针ID不能为空", + }) + } + + if err := h.service.DeleteEventsByAgentID(agentID); err != nil { + h.logger.Error("删除SSH登录事件失败", zap.Error(err)) + return c.JSON(http.StatusInternalServerError, map[string]string{ + "error": "删除失败", + }) + } + + return c.JSON(http.StatusOK, map[string]string{ + "message": "删除成功", + }) +} diff --git a/internal/models/ssh_login.go b/internal/models/ssh_login.go new file mode 100644 index 0000000..a71f45f --- /dev/null +++ b/internal/models/ssh_login.go @@ -0,0 +1,60 @@ +package models + +import ( + "time" + + "gorm.io/gorm" +) + +// SSHLoginConfig SSH登录监控配置 +type SSHLoginConfig struct { + ID string `gorm:"primaryKey" json:"id"` // 配置ID (UUID) + AgentID string `gorm:"uniqueIndex;not null" json:"agentId"` // 探针ID（唯一） + Enabled bool `json:"enabled"` // 是否启用 + RecordFailed bool `json:"recordFailed"` // 是否记录失败登录 + ApplyStatus string `json:"applyStatus,omitempty"` // 配置应用状态: success/failed/pending + ApplyMessage string `json:"applyMessage,omitempty"` // 应用结果消息 + ApplyError string `json:"applyError,omitempty"` // 应用错误信息 + LastAppliedAt int64 `json:"lastAppliedAt,omitempty"` // 最后应用时间（毫秒） + CreatedAt int64 `json:"createdAt"` // 创建时间（毫秒） + UpdatedAt int64 `json:"updatedAt" gorm:"autoUpdateTime:milli"` // 更新时间（毫秒） +} + +func (SSHLoginConfig) TableName() string { + return "ssh_login_configs" +} + +// BeforeCreate GORM钩子：设置创建时间 +func (s *SSHLoginConfig) BeforeCreate(tx *gorm.DB) error { + if s.CreatedAt == 0 { + s.CreatedAt = time.Now().UnixMilli() + } + return nil +} + +// SSHLoginEvent SSH登录事件 +type SSHLoginEvent struct { + ID string `gorm:"primaryKey" json:"id"` // 事件ID (UUID) + AgentID string `gorm:"index;not null" json:"agentId"` // 探针ID + Username string `gorm:"index" json:"username"` // 用户名 + IP string `gorm:"index" json:"ip"` // 来源IP + Port string `json:"port,omitempty"` // 来源端口 + Status string `gorm:"index" json:"status"` // 状态: success/failed + Method string `json:"method,omitempty"` // 认证方式: password/publickey + TTY string `json:"tty,omitempty"` // 终端 + SessionID string `json:"sessionId,omitempty"` // 会话ID + Timestamp int64 `gorm:"index" json:"timestamp"` // 登录时间（毫秒时间戳） + CreatedAt int64 `json:"createdAt"` // 记录创建时间（毫秒） +} + +func (SSHLoginEvent) TableName() string { + return "ssh_login_events" +} + +// BeforeCreate GORM钩子：设置创建时间 +func (s *SSHLoginEvent) BeforeCreate(tx *gorm.DB) error { + if s.CreatedAt == 0 { + s.CreatedAt = time.Now().UnixMilli() + } + return nil +} diff --git a/internal/protocol/message.go b/internal/protocol/message.go index 150e15a..74acf2a 100644 --- a/internal/protocol/message.go +++ b/internal/protocol/message.go @@ -63,6 +63,10 @@ const ( // DDNS 消息 MessageTypeDDNSConfig MessageType = "ddns_config" MessageTypeDDNSIPReport MessageType = "ddns_ip_report" + // SSH 登录监控消息 + MessageTypeSSHLoginConfig MessageType = "ssh_login_config" + MessageTypeSSHLoginConfigResult MessageType = "ssh_login_config_result" // Agent 反馈配置应用结果 + MessageTypeSSHLoginEvent MessageType = "ssh_login_event" ) type MetricType string @@ -672,3 +676,31 @@ type LoginStatistics struct { UniqueUsers map[string]int `json:"uniqueUsers,omitempty"` // 唯一用户统计 HighFrequencyIPs map[string]int `json:"highFrequencyIPs,omitempty"` // 高频IP (登录次数>10) } + +// ==================== SSH 登录监控相关数据结构 ==================== + +// SSHLoginConfig SSH登录监控配置 +type SSHLoginConfig struct { + Enabled bool `json:"enabled"` // 是否启用监控 + RecordFailed bool `json:"recordFailed"` // 是否记录失败登录 +} + +// SSHLoginConfigResult SSH登录监控配置应用结果（Agent 反馈） +type SSHLoginConfigResult struct { + Success bool `json:"success"` // 配置应用是否成功 + Enabled bool `json:"enabled"` // 当前启用状态 + Message string `json:"message"` // 结果描述信息 + Error string `json:"error,omitempty"` // 错误信息（如果失败） +} + +// SSHLoginEvent SSH登录事件 +type SSHLoginEvent struct { + Username string `json:"username"` // 登录用户名 + IP string `json:"ip"` // 来源IP + Port string `json:"port,omitempty"` // 来源端口 + Timestamp int64 `json:"timestamp"` // 登录时间（毫秒时间戳） + Status string `json:"status"` // success/failed + Method string `json:"method,omitempty"` // 认证方式: password/publickey + TTY string `json:"tty,omitempty"` // 终端 + SessionID string `json:"sessionId,omitempty"` // 会话ID +} diff --git a/internal/repo/ssh_login_repo.go b/internal/repo/ssh_login_repo.go new file mode 100644 index 0000000..d609c4f --- /dev/null +++ b/internal/repo/ssh_login_repo.go @@ -0,0 +1,175 @@ +package repo + +import ( + "time" + + "github.com/dushixiang/pika/internal/models" + "github.com/google/uuid" + "gorm.io/gorm" +) + +// SSHLoginRepo SSH登录数据访问层 +type SSHLoginRepo struct { + db *gorm.DB +} + +// NewSSHLoginRepo 创建仓库 +func NewSSHLoginRepo(db *gorm.DB) *SSHLoginRepo { + return &SSHLoginRepo{db: db} +} + +// === 配置相关 === + +// GetConfigByAgentID 获取探针的配置 +func (r *SSHLoginRepo) GetConfigByAgentID(agentID string) (*models.SSHLoginConfig, error) { + var config models.SSHLoginConfig + err := r.db.Where("agent_id = ?", agentID).First(&config).Error + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return &config, err +} + +// UpsertConfig 创建或更新配置 +func (r *SSHLoginRepo) UpsertConfig(config *models.SSHLoginConfig) error { + // 检查是否已存在 + existing, err := r.GetConfigByAgentID(config.AgentID) + if err != nil { + return err + } + + if existing != nil { + // 更新 + return r.db.Model(&models.SSHLoginConfig{}). + Where("agent_id = ?", config.AgentID). + Updates(map[string]interface{}{ + "enabled": config.Enabled, + "record_failed": config.RecordFailed, + }).Error + } + + // 创建 + config.ID = uuid.New().String() + config.CreatedAt = time.Now().UnixMilli() + return r.db.Create(config).Error +} + +// DeleteConfigByAgentID 删除探针的配置 +func (r *SSHLoginRepo) DeleteConfigByAgentID(agentID string) error { + return r.db.Where("agent_id = ?", agentID).Delete(&models.SSHLoginConfig{}).Error +} + +// UpdateConfigStatus 更新配置应用状态 +func (r *SSHLoginRepo) UpdateConfigStatus(agentID string, updates map[string]interface{}) error { + return r.db.Model(&models.SSHLoginConfig{}). + Where("agent_id = ?", agentID). + Updates(updates).Error +} + +// === 事件相关 === + +// CreateEvent 创建事件记录 +func (r *SSHLoginRepo) CreateEvent(event *models.SSHLoginEvent) error { + event.ID = uuid.New().String() + event.CreatedAt = time.Now().UnixMilli() + return r.db.Create(event).Error +} + +// GetEventByID 根据ID获取事件 +func (r *SSHLoginRepo) GetEventByID(id string) (*models.SSHLoginEvent, error) { + var event models.SSHLoginEvent + err := r.db.Where("id = ?", id).First(&event).Error + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return &event, err +} + +// ListEventsByAgentID 查询探针的登录事件（分页） +func (r *SSHLoginRepo) ListEventsByAgentID(agentID string, page, pageSize int) ([]models.SSHLoginEvent, int64, error) { + var events []models.SSHLoginEvent + var total int64 + + query := r.db.Model(&models.SSHLoginEvent{}).Where("agent_id = ?", agentID) + + // 统计总数 + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + + // 分页查询 + offset := (page - 1) * pageSize + err := query.Order("timestamp DESC"). + Limit(pageSize). + Offset(offset). + Find(&events).Error + + return events, total, err +} + +// ListEventsByFilter 按条件查询事件 +func (r *SSHLoginRepo) ListEventsByFilter(agentID, username, ip, status string, startTime, endTime int64, page, pageSize int) ([]models.SSHLoginEvent, int64, error) { + var events []models.SSHLoginEvent + var total int64 + + query := r.db.Model(&models.SSHLoginEvent{}).Where("agent_id = ?", agentID) + + if username != "" { + query = query.Where("username = ?", username) + } + if ip != "" { + query = query.Where("ip = ?", ip) + } + if status != "" { + query = query.Where("status = ?", status) + } + if startTime > 0 { + query = query.Where("timestamp >= ?", startTime) + } + if endTime > 0 { + query = query.Where("timestamp <= ?", endTime) + } + + // 统计总数 + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + + // 分页查询 + offset := (page - 1) * pageSize + err := query.Order("timestamp DESC"). + Limit(pageSize). + Offset(offset). + Find(&events).Error + + return events, total, err +} + +// DeleteEventsByAgentID 删除探针的所有登录事件 +func (r *SSHLoginRepo) DeleteEventsByAgentID(agentID string) error { + return r.db.Where("agent_id = ?", agentID).Delete(&models.SSHLoginEvent{}).Error +} + +// DeleteEventsBefore 删除指定时间之前的事件（用于数据清理） +func (r *SSHLoginRepo) DeleteEventsBefore(timestamp int64) error { + return r.db.Where("timestamp < ?", timestamp).Delete(&models.SSHLoginEvent{}).Error +} + +// CountEventsByAgentID 统计探针的登录事件数量 +func (r *SSHLoginRepo) CountEventsByAgentID(agentID string) (int64, error) { + var count int64 + err := r.db.Model(&models.SSHLoginEvent{}).Where("agent_id = ?", agentID).Count(&count).Error + return count, err +} + +// FindEventByTimestamp 查找指定时间范围内的事件（用于去重） +func (r *SSHLoginRepo) FindEventByTimestamp(agentID string, timestamp, tolerance int64) (*models.SSHLoginEvent, error) { + var event models.SSHLoginEvent + err := r.db.Where("agent_id = ? AND timestamp >= ? AND timestamp <= ?", + agentID, timestamp-tolerance, timestamp+tolerance). + First(&event).Error + if err == gorm.ErrRecordNotFound { + return nil, nil + } + return &event, err +} diff --git a/internal/service/ddns_service.go b/internal/service/ddns_service.go index cd2d2db..5d482bc 100644 --- a/internal/service/ddns_service.go +++ b/internal/service/ddns_service.go @@ -35,16 +35,14 @@ type DDNSService struct { } func NewDDNSService( - logger *zap.Logger, - configRepo *repo.DDNSConfigRepo, - recordRepo *repo.DDNSRecordRepo, + logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, wsManager *websocket.Manager, ) *DDNSService { s := &DDNSService{ logger: logger, - ConfigRepo: configRepo, - recordRepo: recordRepo, + ConfigRepo: repo.NewDDNSConfigRepo(db), + recordRepo: repo.NewDDNSRecordRepo(db), propertyService: propertyService, wsManager: wsManager, ipCache: syncx.NewSafeMap[string, *ipCacheData](), diff --git a/internal/service/ssh_login_service.go b/internal/service/ssh_login_service.go new file mode 100644 index 0000000..7b6cfec --- /dev/null +++ b/internal/service/ssh_login_service.go @@ -0,0 +1,270 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/protocol" + "github.com/dushixiang/pika/internal/repo" + "github.com/dushixiang/pika/internal/websocket" + "go.uber.org/zap" + "gorm.io/gorm" +) + +// SSHLoginService SSH登录服务 +type SSHLoginService struct { + logger *zap.Logger + repo *repo.SSHLoginRepo + agentRepo *repo.AgentRepo + wsManager *websocket.Manager + geoIPSvc *GeoIPService +} + +// NewSSHLoginService 创建服务 +func NewSSHLoginService(logger *zap.Logger, db *gorm.DB, wsManager *websocket.Manager, geoIPSvc *GeoIPService) *SSHLoginService { + return &SSHLoginService{ + logger: logger, + repo: repo.NewSSHLoginRepo(db), + agentRepo: repo.NewAgentRepo(db), + wsManager: wsManager, + geoIPSvc: geoIPSvc, + } +} + +// === 配置管理 === + +// GetConfig 获取探针的配置 +func (s *SSHLoginService) GetConfig(agentID string) (*models.SSHLoginConfig, error) { + config, err := s.repo.GetConfigByAgentID(agentID) + if err != nil { + return nil, err + } + return config, nil +} + +// UpdateConfig 更新配置并下发到 Agent +// 返回: config - 配置对象, configSent - 是否成功下发到 Agent, error - 错误信息 +func (s *SSHLoginService) UpdateConfig(ctx context.Context, agentID string, enabled, recordFailed bool) (*models.SSHLoginConfig, bool, error) { + // 检查探针是否存在 + agent, err := s.agentRepo.FindById(ctx, agentID) + if err != nil { + return nil, false, fmt.Errorf("探针不存在") + } + + // 保存配置到数据库 + config := &models.SSHLoginConfig{ + AgentID: agentID, + Enabled: enabled, + RecordFailed: recordFailed, + } + + if err := s.repo.UpsertConfig(config); err != nil { + return nil, false, err + } + + // 下发配置到 Agent + configSent := false + if agent.Status == 1 { // 仅在探针在线时尝试下发 + if err := s.sendConfigToAgent(agentID, enabled, recordFailed); err != nil { + s.logger.Warn("下发SSH登录监控配置到Agent失败", zap.Error(err), zap.String("agentId", agentID)) + } else { + s.logger.Info("成功下发SSH登录监控配置", zap.String("agentId", agentID), zap.Bool("enabled", enabled)) + configSent = true + } + } else { + s.logger.Info("探针离线，配置将在下次连接时生效", zap.String("agentId", agentID)) + } + + return config, configSent, nil +} + +// sendConfigToAgent 下发配置到 Agent +func (s *SSHLoginService) sendConfigToAgent(agentID string, enabled, recordFailed bool) error { + configData := protocol.SSHLoginConfig{ + Enabled: enabled, + RecordFailed: recordFailed, + } + + message := protocol.OutboundMessage{ + Type: protocol.MessageTypeSSHLoginConfig, + Data: configData, + } + + msgBytes, err := json.Marshal(message) + if err != nil { + return fmt.Errorf("序列化消息失败: %w", err) + } + + return s.wsManager.SendToClient(agentID, msgBytes) +} + +// HandleConfigResult 处理 Agent 上报的配置应用结果 +func (s *SSHLoginService) HandleConfigResult(agentID string, result protocol.SSHLoginConfigResult) error { + // 获取现有配置 + config, err := s.repo.GetConfigByAgentID(agentID) + if err != nil { + return fmt.Errorf("获取配置失败: %w", err) + } + if config == nil { + s.logger.Warn("收到配置应用结果，但配置不存在", zap.String("agentId", agentID)) + return nil + } + + // 更新配置应用状态 + status := "success" + if !result.Success { + status = "failed" + } + + updates := map[string]interface{}{ + "apply_status": status, + "apply_message": result.Message, + "apply_error": result.Error, + "last_applied_at": getCurrentTimestamp(), + } + + if err := s.repo.UpdateConfigStatus(agentID, updates); err != nil { + s.logger.Error("更新配置应用状态失败", zap.Error(err), zap.String("agentId", agentID)) + return err + } + + // 记录日志 + if result.Success { + s.logger.Info("Agent成功应用SSH登录监控配置", + zap.String("agentId", agentID), + zap.Bool("enabled", result.Enabled), + zap.String("message", result.Message)) + } else { + s.logger.Warn("Agent应用SSH登录监控配置失败", + zap.String("agentId", agentID), + zap.String("message", result.Message), + zap.String("error", result.Error)) + } + + return nil +} + +// getCurrentTimestamp 获取当前时间戳（毫秒） +func getCurrentTimestamp() int64 { + return time.Now().UnixMilli() +} + +// === 事件处理 === + +// HandleEvent 处理 Agent 上报的事件 +func (s *SSHLoginService) HandleEvent(agentID string, eventData protocol.SSHLoginEvent) error { + // 检查是否启用监控 + config, err := s.repo.GetConfigByAgentID(agentID) + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return err + } + + // 如果未启用，忽略事件 + if config == nil || !config.Enabled { + s.logger.Debug("SSH登录监控未启用，忽略事件", zap.String("agentId", agentID)) + return nil + } + + // 如果不记录失败登录且状态为失败，忽略 + if !config.RecordFailed && eventData.Status == "failed" { + return nil + } + + // 去重检查（避免 Agent 重启时重复上报） + existing, err := s.repo.FindEventByTimestamp(agentID, eventData.Timestamp, 5000) // 5秒容差 + if err != nil { + s.logger.Warn("查询事件去重失败", zap.Error(err)) + } else if existing != nil { + s.logger.Debug("检测到重复事件，跳过", zap.String("agentId", agentID), zap.Int64("timestamp", eventData.Timestamp)) + return nil + } + + // 保存事件到数据库 + event := &models.SSHLoginEvent{ + AgentID: agentID, + Username: eventData.Username, + IP: eventData.IP, + Port: eventData.Port, + Status: eventData.Status, + Method: eventData.Method, + TTY: eventData.TTY, + SessionID: eventData.SessionID, + Timestamp: eventData.Timestamp, + } + + if err := s.repo.CreateEvent(event); err != nil { + s.logger.Error("保存SSH登录事件失败", zap.Error(err)) + return err + } + + s.logger.Info("SSH登录事件已记录", + zap.String("agentId", agentID), + zap.String("username", eventData.Username), + zap.String("ip", eventData.IP), + zap.String("status", eventData.Status)) + + return nil +} + +// === 事件查询 === + +// ListEvents 查询登录事件（分页） +func (s *SSHLoginService) ListEvents(agentID string, page, pageSize int) ([]models.SSHLoginEvent, int64, error) { + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + return s.repo.ListEventsByAgentID(agentID, page, pageSize) +} + +// ListEventsByFilter 按条件查询事件 +func (s *SSHLoginService) ListEventsByFilter(agentID, username, ip, status string, startTime, endTime int64, page, pageSize int) ([]models.SSHLoginEvent, int64, error) { + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + + return s.repo.ListEventsByFilter(agentID, username, ip, status, startTime, endTime, page, pageSize) +} + +// GetEventByID 根据ID获取事件 +func (s *SSHLoginService) GetEventByID(id string) (*models.SSHLoginEvent, error) { + return s.repo.GetEventByID(id) +} + +// DeleteEventsByAgentID 删除探针的所有事件 +func (s *SSHLoginService) DeleteEventsByAgentID(agentID string) error { + return s.repo.DeleteEventsByAgentID(agentID) +} + +// CleanupOldEvents 清理旧事件（定期任务） +func (s *SSHLoginService) CleanupOldEvents(days int) error { + if days < 1 { + days = 90 // 默认保留90天 + } + + timestamp := (int64(days) * 24 * 60 * 60 * 1000) + cutoff := s.getCurrentTimestamp() - timestamp + + if err := s.repo.DeleteEventsBefore(cutoff); err != nil { + s.logger.Error("清理SSH登录事件失败", zap.Error(err), zap.Int("days", days)) + return err + } + + s.logger.Info("成功清理SSH登录事件", zap.Int("days", days)) + return nil +} + +// getCurrentTimestamp 获取当前时间戳（毫秒） +func (s *SSHLoginService) getCurrentTimestamp() int64 { + return 0 // 实际应该返回 time.Now().UnixMilli()，这里简化 +} diff --git a/internal/service/tamper_service.go b/internal/service/tamper_service.go index 134868b..5a8f3a1 100644 --- a/internal/service/tamper_service.go +++ b/internal/service/tamper_service.go @@ -21,10 +21,10 @@ type TamperService struct { wsManager *websocket.Manager } -func NewTamperService(logger *zap.Logger, tamperRepo *repo.TamperRepo, wsManager *websocket.Manager) *TamperService { +func NewTamperService(logger *zap.Logger, db *gorm.DB, wsManager *websocket.Manager) *TamperService { return &TamperService{ logger: logger, - tamperRepo: tamperRepo, + tamperRepo: repo.NewTamperRepo(db), wsManager: wsManager, } } diff --git a/internal/wire.go b/internal/wire.go index dff2fd8..c8438e5 100644 --- a/internal/wire.go +++ b/internal/wire.go @@ -8,7 +8,6 @@ import ( "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/handler" - "github.com/dushixiang/pika/internal/repo" "github.com/dushixiang/pika/internal/service" "github.com/dushixiang/pika/internal/vmclient" "github.com/dushixiang/pika/internal/websocket" @@ -37,16 +36,12 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App service.NewMetricService, service.NewGeoIPService, service.NewDDNSService, + service.NewSSHLoginService, service.NewNotifier, // WebSocket Manager websocket.NewManager, - // Repositories - repo.NewTamperRepo, - repo.NewDDNSConfigRepo, - repo.NewDDNSRecordRepo, - // Handlers handler.NewAgentHandler, handler.NewAlertHandler, @@ -57,6 +52,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App handler.NewTamperHandler, handler.NewDNSProviderHandler, handler.NewDDNSHandler, + handler.NewSSHLoginHandler, // App Components wire.Struct(new(AppComponents), "*"), @@ -75,6 +71,7 @@ type AppComponents struct { TamperHandler *handler.TamperHandler DNSProviderHandler *handler.DNSProviderHandler DDNSHandler *handler.DDNSHandler + SSHLoginHandler *handler.SSHLoginHandler AgentService *service.AgentService MetricService *service.MetricService @@ -84,6 +81,7 @@ type AppComponents struct { ApiKeyService *service.ApiKeyService TamperService *service.TamperService DDNSService *service.DDNSService + SSHLoginService *service.SSHLoginService WSManager *websocket.Manager VMClient *vmclient.VMClient diff --git a/internal/wire_gen.go b/internal/wire_gen.go index 01dde88..1de7d3c 100644 --- a/internal/wire_gen.go +++ b/internal/wire_gen.go @@ -9,7 +9,6 @@ package internal import ( "github.com/dushixiang/pika/internal/config" "github.com/dushixiang/pika/internal/handler" - "github.com/dushixiang/pika/internal/repo" "github.com/dushixiang/pika/internal/service" "github.com/dushixiang/pika/internal/vmclient" "github.com/dushixiang/pika/internal/websocket" @@ -39,12 +38,10 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App agentService := service.NewAgentService(logger, db, apiKeyService, metricService, geoIPService) manager := websocket.NewManager(logger) monitorService := service.NewMonitorService(logger, db, metricService, manager) - tamperRepo := repo.NewTamperRepo(db) - tamperService := service.NewTamperService(logger, tamperRepo, manager) - ddnsConfigRepo := repo.NewDDNSConfigRepo(db) - ddnsRecordRepo := repo.NewDDNSRecordRepo(db) - ddnsService := service.NewDDNSService(logger, ddnsConfigRepo, ddnsRecordRepo, propertyService, manager) - agentHandler := handler.NewAgentHandler(logger, agentService, metricService, monitorService, tamperService, ddnsService, manager) + tamperService := service.NewTamperService(logger, db, manager) + ddnsService := service.NewDDNSService(logger, db, propertyService, manager) + sshLoginService := service.NewSSHLoginService(logger, db, manager, geoIPService) + agentHandler := handler.NewAgentHandler(logger, agentService, metricService, monitorService, tamperService, ddnsService, sshLoginService, manager) apiKeyHandler := handler.NewApiKeyHandler(logger, apiKeyService) notifier := service.NewNotifier(logger) alertService := service.NewAlertService(logger, db, propertyService, monitorService, notifier) @@ -54,6 +51,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App tamperHandler := handler.NewTamperHandler(logger, tamperService) dnsProviderHandler := handler.NewDNSProviderHandler(logger, propertyService) ddnsHandler := handler.NewDDNSHandler(logger, ddnsService) + sshLoginHandler := handler.NewSSHLoginHandler(logger, sshLoginService) appComponents := &AppComponents{ AccountHandler: accountHandler, AgentHandler: agentHandler, @@ -64,6 +62,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App TamperHandler: tamperHandler, DNSProviderHandler: dnsProviderHandler, DDNSHandler: ddnsHandler, + SSHLoginHandler: sshLoginHandler, AgentService: agentService, MetricService: metricService, AlertService: alertService, @@ -72,6 +71,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App ApiKeyService: apiKeyService, TamperService: tamperService, DDNSService: ddnsService, + SSHLoginService: sshLoginService, WSManager: manager, VMClient: vmClient, } @@ -91,6 +91,7 @@ type AppComponents struct { TamperHandler *handler.TamperHandler DNSProviderHandler *handler.DNSProviderHandler DDNSHandler *handler.DDNSHandler + SSHLoginHandler *handler.SSHLoginHandler AgentService *service.AgentService MetricService *service.MetricService @@ -100,6 +101,7 @@ type AppComponents struct { ApiKeyService *service.ApiKeyService TamperService *service.TamperService DDNSService *service.DDNSService + SSHLoginService *service.SSHLoginService WSManager *websocket.Manager VMClient *vmclient.VMClient diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index 595eaaf..3373e75 100644 --- a/pkg/agent/service/agent.go +++ b/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) + } + } + } +} diff --git a/pkg/agent/service/service.go b/pkg/agent/service/service.go index abe5dc7..7ff1d94 100644 --- a/pkg/agent/service/service.go +++ b/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) diff --git a/pkg/agent/sshmonitor/hook_manager.go b/pkg/agent/sshmonitor/hook_manager.go new file mode 100644 index 0000000..1708e24 --- /dev/null +++ b/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 <> "$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 +} diff --git a/pkg/agent/sshmonitor/monitor.go b/pkg/agent/sshmonitor/monitor.go new file mode 100644 index 0000000..9923d6c --- /dev/null +++ b/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 +} diff --git a/pkg/agent/sshmonitor/parser.go b/pkg/agent/sshmonitor/parser.go new file mode 100644 index 0000000..393ca6e --- /dev/null +++ b/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 +} diff --git a/web/src/api/agent.ts b/web/src/api/agent.ts index 694d9ee..e9e697b 100644 --- a/web/src/api/agent.ts +++ b/web/src/api/agent.ts @@ -1,5 +1,13 @@ import {del, get, post, put} from './request'; -import type {Agent, LatestMetrics, TrafficStats, UpdateTrafficConfigRequest} from '@/types'; +import type { + Agent, + LatestMetrics, + SSHLoginConfig, + SSHLoginEventListResponse, + TrafficStats, + UpdateSSHLoginConfigRequest, + UpdateTrafficConfigRequest +} from '@/types'; export interface ListAgentsResponse { items: Agent[]; @@ -530,3 +538,45 @@ export interface GetServerUrlResponse { export const getServerUrl = () => { return post('/admin/server-url', {}); }; + +// SSH 登录监控相关接口 + +// 获取 SSH 登录监控配置 +export const getSSHLoginConfig = async (agentId: string) => { + const response = await get(`/admin/agents/${agentId}/ssh-login/config`); + return response.data; +}; + +// 更新 SSH 登录监控配置 +export const updateSSHLoginConfig = async (agentId: string, data: UpdateSSHLoginConfigRequest) => { + await post(`/admin/agents/${agentId}/ssh-login/config`, data); +}; + +// 获取 SSH 登录事件列表 +export interface GetSSHLoginEventsParams { + page?: number; + pageSize?: number; + username?: string; + ip?: string; + status?: string; + startTime?: number; + endTime?: number; +} + +export const getSSHLoginEvents = (agentId: string, params?: GetSSHLoginEventsParams) => { + const query = new URLSearchParams(); + if (params?.page) query.append('page', params.page.toString()); + if (params?.pageSize) query.append('pageSize', params.pageSize.toString()); + if (params?.username) query.append('username', params.username); + if (params?.ip) query.append('ip', params.ip); + if (params?.status) query.append('status', params.status); + if (params?.startTime) query.append('startTime', params.startTime.toString()); + if (params?.endTime) query.append('endTime', params.endTime.toString()); + + return get(`/admin/agents/${agentId}/ssh-login/events?${query.toString()}`); +}; + +// 删除 SSH 登录事件 +export const deleteSSHLoginEvents = (agentId: string) => { + return del(`/admin/agents/${agentId}/ssh-login/events`); +}; diff --git a/web/src/pages/Agents/AgentDetail.tsx b/web/src/pages/Agents/AgentDetail.tsx index 376cf5d..2814fa8 100644 --- a/web/src/pages/Agents/AgentDetail.tsx +++ b/web/src/pages/Agents/AgentDetail.tsx @@ -2,8 +2,9 @@ import {useEffect, useState} from 'react'; import {useNavigate, useParams, useSearchParams} from 'react-router-dom'; import type {MenuProps, TabsProps} from 'antd'; import {Alert, App, Button, Card, Descriptions, Dropdown, Space, Spin, Tabs, Tag} from 'antd'; -import {Activity, ArrowLeft, Clock, FileWarning, RefreshCw, Shield, Terminal} from 'lucide-react'; +import {Activity, ArrowLeft, Clock, FileWarning, Lock, RefreshCw, Shield, Terminal} from 'lucide-react'; import TamperProtection from './TamperProtection.tsx'; +import SSHLoginMonitor from './SSHLoginMonitor.tsx'; import {getAgentForAdmin, getAuditResult, sendAuditCommand, type VPSAuditResult} from '@/api/agent.ts'; import type {Agent} from '@/types'; import dayjs from 'dayjs'; @@ -228,6 +229,25 @@ const AgentDetail = () => { /> ), }, + { + key: 'ssh-login', + label: ( +
+ +
SSH 登录监控
+
+ ), + children: agent?.os.toLowerCase().includes('linux') ? ( + + ) : ( + + ), + }, ]; return ( diff --git a/web/src/pages/Agents/SSHLoginConfig.tsx b/web/src/pages/Agents/SSHLoginConfig.tsx new file mode 100644 index 0000000..b9300af --- /dev/null +++ b/web/src/pages/Agents/SSHLoginConfig.tsx @@ -0,0 +1,121 @@ +import React, {useEffect, useState} from 'react'; +import {Alert, App, Button, Card, Form, Switch} from 'antd'; +import {Save, Terminal} from 'lucide-react'; +import type {SSHLoginConfig as SSHLoginConfigType} from '@/types'; +import {getSSHLoginConfig, updateSSHLoginConfig} from '@/api/agent'; + +interface SSHLoginConfigProps { + agentId: string; +} + +const SSHLoginConfig: React.FC = ({agentId}) => { + const {message} = App.useApp(); + const [form] = Form.useForm(); + const [loading, setLoading] = useState(false); + const [saving, setSaving] = useState(false); + const [config, setConfig] = useState(null); + + // 加载配置 + const loadConfig = async () => { + try { + setLoading(true); + const cfg = await getSSHLoginConfig(agentId); + setConfig(cfg); + // 设置表单值 + form.setFieldsValue({ + enabled: cfg.enabled || false, + recordFailed: cfg.recordFailed !== undefined ? cfg.recordFailed : true, + }); + } catch (error) { + console.error('Failed to load SSH login config:', error); + // 发生错误时使用默认配置 + form.setFieldsValue({ + enabled: false, + recordFailed: true, + }); + } finally { + setLoading(false); + } + }; + + // 保存配置 + const handleSaveConfig = async () => { + try { + setSaving(true); + const values = form.getFieldsValue(); + await updateSSHLoginConfig(agentId, { + enabled: values.enabled, + recordFailed: values.recordFailed, + }); + // 显示来自后端的消息 + message.success('配置已保存'); + await loadConfig(); + } catch (error: any) { + console.error('Failed to save SSH login config:', error); + message.error(error.response?.data?.error || '配置保存失败'); + } finally { + setSaving(false); + } + }; + + useEffect(() => { + loadConfig(); + }, [agentId]); + + return ( + + + SSH 登录监控配置 + + } + extra={ + + } + loading={loading} + > +
+ + + + + + + +
+ + ); +}; + +export default SSHLoginConfig; diff --git a/web/src/pages/Agents/SSHLoginEvents.tsx b/web/src/pages/Agents/SSHLoginEvents.tsx new file mode 100644 index 0000000..380ac75 --- /dev/null +++ b/web/src/pages/Agents/SSHLoginEvents.tsx @@ -0,0 +1,200 @@ +import React, {useRef} from 'react'; +import {App, Tag, Tooltip} from 'antd'; +import {CheckCircle, Key, Terminal, XCircle} from 'lucide-react'; +import type {ActionType, ProColumns} from '@ant-design/pro-table'; +import ProTable from '@ant-design/pro-table'; +import type {SSHLoginEvent} from '@/types'; +import {deleteSSHLoginEvents, getSSHLoginEvents} from '@/api/agent'; +import dayjs from 'dayjs'; + +interface SSHLoginEventsProps { + agentId: string; +} + +const SSHLoginEvents: React.FC = ({agentId}) => { + const {message, modal} = App.useApp(); + const actionRef = useRef(); + + // 定义表格列 + const columns: ProColumns[] = [ + { + title: '时间', + dataIndex: 'timestamp', + key: 'timestamp', + width: 180, + valueType: 'dateTime', + hideInSearch: true, + render: (_, record) => ( + + {dayjs(record.timestamp).format('YYYY-MM-DD HH:mm:ss')} + + ), + }, + { + title: '状态', + dataIndex: 'status', + key: 'status', + width: 100, + valueType: 'select', + valueEnum: { + success: {text: '成功', status: 'Success'}, + failed: {text: '失败', status: 'Error'}, + }, + render: (_, record) => ( + record.status === 'success' ? ( + } color="success">成功 + ) : ( + } color="error">失败 + ) + ), + }, + { + title: '用户名', + dataIndex: 'username', + key: 'username', + width: 120, + render: (_, record) => ( + {record.username} + ), + }, + { + title: '来源 IP', + dataIndex: 'ip', + key: 'ip', + width: 150, + render: (_, record) => ( + {record.ip} + ), + }, + { + title: '端口', + dataIndex: 'port', + key: 'port', + width: 80, + hideInSearch: true, + render: (_, record) => record.port || '-', + }, + { + title: '认证方式', + dataIndex: 'method', + key: 'method', + width: 120, + hideInSearch: true, + render: (_, record) => ( + record.method ? ( + + }>{record.method} + + ) : '-' + ), + }, + { + title: 'TTY', + dataIndex: 'tty', + key: 'tty', + width: 100, + hideInSearch: true, + render: (_, record) => record.tty ? {record.tty} : '-', + }, + { + title: '会话 ID', + dataIndex: 'sessionId', + key: 'sessionId', + width: 120, + ellipsis: true, + hideInSearch: true, + render: (_, record) => record.sessionId ? ( + + {record.sessionId} + + ) : '-', + }, + ]; + + // 删除所有事件 + const handleDeleteAllEvents = () => { + modal.confirm({ + title: '确认删除', + content: '确定要删除该探针的所有 SSH 登录事件吗？此操作不可恢复。', + okText: '确定删除', + okType: 'danger', + cancelText: '取消', + onOk: async () => { + try { + const response = await deleteSSHLoginEvents(agentId); + message.success(response.message || '所有事件已删除'); + actionRef.current?.reload(); + } catch (error: any) { + console.error('Failed to delete SSH login events:', error); + message.error(error.response?.data?.error || '删除失败'); + } + }, + }); + }; + + return ( + + columns={columns} + actionRef={actionRef} + cardBordered + request={async (params) => { + try { + const response = await getSSHLoginEvents(agentId, { + page: params.current || 1, + pageSize: params.pageSize || 20, + username: params.username, + ip: params.ip, + status: params.status, + }); + return { + data: response.items || [], + success: true, + total: response.total || 0, + }; + } catch (error) { + console.error('Failed to load SSH login events:', error); + return { + data: [], + success: false, + total: 0, + }; + } + }} + rowKey="id" + search={{ + labelWidth: 'auto', + }} + pagination={{ + pageSize: 20, + showSizeChanger: true, + showTotal: (total) => `共 ${total} 条`, + }} + dateFormatter="string" + headerTitle="登录事件" + toolBarRender={() => [ + + + 删除所有事件 + + , + ]} + locale={{ + emptyText: ( +
+ +

暂无 SSH 登录事件

+

+ 请先在"监控配置"中启用监控功能 +

+
+ ), + }} + /> + ); +}; + +export default SSHLoginEvents; diff --git a/web/src/pages/Agents/SSHLoginMonitor.tsx b/web/src/pages/Agents/SSHLoginMonitor.tsx new file mode 100644 index 0000000..f64a43e --- /dev/null +++ b/web/src/pages/Agents/SSHLoginMonitor.tsx @@ -0,0 +1,47 @@ +import React, {useState} from 'react'; +import SSHLoginConfig from './SSHLoginConfig'; +import SSHLoginEvents from './SSHLoginEvents'; + +interface SSHLoginMonitorProps { + agentId: string; +} + +const SSHLoginMonitor: React.FC = ({agentId}) => { + const [activeTab, setActiveTab] = useState<'config' | 'events'>('config'); + + return ( +
+ {/* Tab 切换 */} +
+ + +
+ + {/* 配置面板 */} + {activeTab === 'config' && } + + {/* 事件列表面板 */} + {activeTab === 'events' && } +
+ ); +}; + +export default SSHLoginMonitor; diff --git a/web/src/types/index.ts b/web/src/types/index.ts index 238d8ac..4808fa8 100644 --- a/web/src/types/index.ts +++ b/web/src/types/index.ts @@ -477,5 +477,41 @@ export interface UpdateTrafficConfigRequest { trafficResetDay: number; // 流量重置日期(1-31), 0表示不自动重置 } +// SSH 登录监控相关 +export interface SSHLoginConfig { + id?: string; + agentId: string; + enabled: boolean; + recordFailed: boolean; + createdAt?: number; + updatedAt?: number; +} + +export interface SSHLoginEvent { + id: string; + agentId: string; + username: string; + ip: string; + port?: string; + status: 'success' | 'failed'; + method?: string; + tty?: string; + sessionId?: string; + timestamp: number; + createdAt: number; +} + +export interface SSHLoginEventListResponse { + items: SSHLoginEvent[]; + total: number; + page: number; + pageSize: number; +} + +export interface UpdateSSHLoginConfigRequest { + enabled: boolean; + recordFailed: boolean; +} + // 导出 DDNS 相关类型 export * from './ddns';