audit

README

Audit 审计模块

概述

审计模块负责记录和管理代理连接的所有审计数据，包括连接审计、数据流审计、操作事件审计和会话管理。

设计原则

1. 解耦设计: 与 proxy 模块解耦，通过接口调用
2. 协议无关: 支持多种协议（RDP/MySQL/SSH）的审计
3. 分层审计:
   • 连接层：连接建立/关闭
   • 数据流层：原始数据包
   • 操作层：高级语义事件（登录、命令、查询等）
4. 可扩展: 为后续协议审计、会话回放预留接口

模块结构

server/apps/cmdb/audit/
├── interface.go          # 审计服务接口定义
├── types.go              # 审计数据结构
├── impl/
│   └── impl.go          # 服务实现（Phase 1: Noop）
└── README.md

接口定义

核心接口

type Service interface {
    // 连接审计
    RecordConnection(context.Context, *ConnectionEvent) error
    CloseConnection(context.Context, *ConnectionCloseEvent) error
    
    // 数据流审计
    RecordDataFlow(context.Context, *DataFlowEvent) error
    
    // 操作事件审计
    RecordEvent(context.Context, *OperationEvent) error
    
    // 会话管理
    GetSession(context.Context, *GetSessionRequest) (*Session, error)
    QuerySessions(context.Context, *QuerySessionRequest) (*SessionSet, error)
    
    // 会话回放（预留）
    GetSessionReplay(context.Context, *GetSessionReplayRequest) (*SessionReplay, error)
}

审计事件类型

1. 连接事件

• connection_open: 连接建立
• connection_close: 连接关闭

2. 数据流事件

• data_send: 数据发送
• data_recv: 数据接收

3. SSH 操作事件

• ssh_login: SSH 登录
• ssh_command: 命令执行
• ssh_file_upload: 文件上传
• ssh_file_download: 文件下载

4. MySQL 操作事件

• mysql_login: MySQL 登录
• mysql_query: 查询操作
• mysql_ddl: DDL 操作（CREATE/ALTER/DROP）
• mysql_dml: DML 操作（INSERT/UPDATE/DELETE）

5. RDP 操作事件

• rdp_login: RDP 登录
• rdp_screenshot: 屏幕截图
• rdp_file_transfer: 文件传输

使用示例

在 Proxy 模块中调用

import "github.com/infraboard/devops/server/apps/mproxy/audit"

// 获取审计服务
auditService := audit.GetService()

// 记录连接建立
err := auditService.RecordConnection(ctx, &audit.ConnectionEvent{
    SessionID:   sessionID,
    Protocol:    audit.ProtocolSSH,
    SourceIP:    clientIP,
    TargetHost:  targetHost,
    Username:    username,
    ConnectTime: time.Now(),
})

// 记录操作事件
err = auditService.RecordEvent(ctx, &audit.OperationEvent{
    SessionID: sessionID,
    Protocol:  audit.ProtocolSSH,
    EventType: audit.EventTypeSSHCommand,
    Operation: "ls -la /home",
    Result:    "success",
    Timestamp: time.Now(),
})

实现阶段

Phase 1: Noop 实现（当前）

• 接口定义完整
• Noop 实现（只记录日志，不持久化）
• Proxy 模块可以正常调用，不影响功能

Phase 2: 基础实现（未来）

• 数据库存储（连接记录、事件记录）
• 基础查询功能
• 会话管理

Phase 3: 协议审计（未来）

• SSH 协议解析和审计
• MySQL 协议解析和审计
• RDP 协议解析和审计
• 会话回放功能

配置示例

[audit.storage]
type = "database"  # database/elasticsearch/file
dsn = "postgres://user:pass@localhost:5432/audit?sslmode=disable"
retention_days = 90

[audit.buffer]
enabled = true
size = 1000
flush_interval = 10  # 秒

数据库设计（预留）

sessions 表

CREATE TABLE audit_sessions (
    id VARCHAR(64) PRIMARY KEY,
    protocol VARCHAR(20),
    source_ip VARCHAR(45),
    source_port INT,
    resource_id VARCHAR(64),
    target_host VARCHAR(255),
    target_port INT,
    username VARCHAR(100),
    user_id VARCHAR(64),
    start_time TIMESTAMP,
    end_time TIMESTAMP,
    duration BIGINT,
    status VARCHAR(20),
    bytes_sent BIGINT,
    bytes_received BIGINT,
    event_count BIGINT,
    metadata JSONB
);

events 表

CREATE TABLE audit_events (
    id BIGSERIAL PRIMARY KEY,
    session_id VARCHAR(64),
    protocol VARCHAR(20),
    event_type VARCHAR(50),
    timestamp TIMESTAMP,
    username VARCHAR(100),
    operation TEXT,
    result VARCHAR(20),
    error_message TEXT,
    data JSONB,
    FOREIGN KEY (session_id) REFERENCES audit_sessions(id)
);

后续扩展方向

1. 实时告警: 基于审计事件的实时告警
2. 行为分析: 用户行为分析和异常检测
3. 合规报告: 自动生成合规审计报告
4. 回放功能: SSH 会话回放、MySQL 操作回放
5. 全文搜索: 基于 Elasticsearch 的全文搜索

Documentation

Index

• Constants
• type ConnectionCloseEvent
• type ConnectionEvent
• type DataFlowEvent
• type EventType
• type GetSessionReplayRequest
• type GetSessionRequest
• type ObjectStorage
• type OperationEvent
  • func (*OperationEvent) TableName() string
• type Protocol
• type QuerySessionRequest
  • func (r *QuerySessionRequest) ComputeOffset() int64
• type Service
  • func GetService() Service
• type Session
  • func (*Session) TableName() string
• type SessionReplay
• type SessionSet

Constants

const (
	AppName = "proxy_audit"
)

Types

type ConnectionCloseEvent

type ConnectionCloseEvent struct {
	// 会话ID
	SessionID string `json:"session_id"`
	// 关闭时间
	CloseTime time.Time `json:"close_time"`
	// 关闭原因
	CloseReason string `json:"close_reason"`
	// 连接时长（秒）
	Duration int64 `json:"duration"`
	// 数据统计
	BytesSent     int64 `json:"bytes_sent"`
	BytesReceived int64 `json:"bytes_received"`
}

ConnectionCloseEvent 连接关闭事件

type ConnectionEvent

type ConnectionEvent struct {
	// 会话ID
	SessionID string `json:"session_id"`
	// 协议类型
	Protocol Protocol `json:"protocol"`
	// 源IP
	SourceIP string `json:"source_ip"`
	// 源端口
	SourcePort int `json:"source_port"`
	// 目标资源ID
	ResourceID string `json:"resource_id"`
	// 目标主机
	TargetHost string `json:"target_host"`
	// 目标端口
	TargetPort int `json:"target_port"`
	// 用户信息
	Username string `json:"username"`
	UserID   string `json:"user_id"`
	// 连接时间
	ConnectTime time.Time `json:"connect_time"`
	// 扩展信息
	Metadata map[string]string `json:"metadata,omitempty"`
}

ConnectionEvent 连接建立事件

type DataFlowEvent

type DataFlowEvent struct {
	// 会话ID
	SessionID string `json:"session_id"`
	// 协议类型
	Protocol Protocol `json:"protocol"`
	// 事件类型
	EventType EventType `json:"event_type"`
	// 时间戳
	Timestamp time.Time `json:"timestamp"`
	// 方向（send/recv）
	Direction string `json:"direction"`
	// 数据大小（字节）
	Size int64 `json:"size"`
	// 协议相关数据
	ProtocolData interface{} `json:"protocol_data,omitempty"`
}

DataFlowEvent 数据流事件（协议解析后的数据）

type EventType

type EventType string

EventType 事件类型

const (
	// 连接事件
	EventTypeConnectionOpen  EventType = "connection_open"
	EventTypeConnectionClose EventType = "connection_close"

	// 数据流事件
	EventTypeDataSend EventType = "data_send"
	EventTypeDataRecv EventType = "data_recv"

	// 操作事件（SSH）
	EventTypeSSHLogin        EventType = "ssh_login"
	EventTypeSSHCommand      EventType = "ssh_command"
	EventTypeSSHFileUpload   EventType = "ssh_file_upload"
	EventTypeSSHFileDownload EventType = "ssh_file_download"

	// 操作事件（MySQL）
	EventTypeMySQLLogin EventType = "mysql_login"
	EventTypeMySQLQuery EventType = "mysql_query"
	EventTypeMySQLDDL   EventType = "mysql_ddl"
	EventTypeMySQLDML   EventType = "mysql_dml"

	// 操作事件（RDP）
	EventTypeRDPLogin        EventType = "rdp_login"
	EventTypeRDPLogout       EventType = "rdp_logout"
	EventTypeRDPScreenshot   EventType = "rdp_screenshot"
	EventTypeRDPFileTransfer EventType = "rdp_file_transfer"

	// 会话事件
	EventTypeSessionClose EventType = "session_close"
)

type GetSessionReplayRequest

type GetSessionReplayRequest struct {
	SessionID string `json:"session_id"`
	// 回放格式（根据协议不同）
	Format string `json:"format,omitempty"`
	// URL 有效期（秒），默认 3600
	URLExpiry int `json:"url_expiry,omitempty"`
}

GetSessionReplayRequest 获取会话回放请求

type GetSessionRequest

type GetSessionRequest struct {
	SessionID string `json:"session_id"`
}

GetSessionRequest 获取会话请求

type ObjectStorage

type ObjectStorage interface {
	// Upload 上传文件
	Upload(ctx context.Context, path string, data io.Reader) error

	// GetURL 获取文件下载 URL
	GetURL(ctx context.Context, path string, expiry time.Duration) (string, error)

	// Download 下载文件（用于分片合并）
	Download(ctx context.Context, path string) ([]byte, error)

	// Delete 删除文件（用于清理分片）
	Delete(ctx context.Context, path string) error

	// Exists 检查文件是否存在
	Exists(ctx context.Context, path string) (bool, error)

	// ListWithPrefix 列出指定前缀的所有对象路径
	ListWithPrefix(ctx context.Context, prefix string) ([]string, error)

	// BatchDownloadToDir 批量下载文件到本地目录
	BatchDownloadToDir(ctx context.Context, paths []string, destDir string) error
}

ObjectStorage 对象存储接口（用于录像文件上传）

type OperationEvent

type OperationEvent struct {
	// 事件ID
	ID int64 `json:"id" gorm:"column:id;primaryKey;autoIncrement"`
	// 会话ID
	SessionID string `json:"session_id" gorm:"column:session_id;size:64;index"`
	// 协议类型
	Protocol Protocol `json:"protocol" gorm:"column:protocol;size:32;index"`
	// 事件类型
	EventType EventType `json:"event_type" gorm:"column:event_type;size:64;index"`
	// 时间戳
	Timestamp int64 `json:"timestamp" gorm:"column:timestamp;index"`
	// 用户信息
	Username string `json:"username" gorm:"column:username;size:100"`
	UserID   string `json:"user_id" gorm:"column:user_id;size:64"`
	// 操作内容
	Operation string `json:"operation" gorm:"column:operation;type:text"`
	// 操作结果（success/failed）
	Result string `json:"result" gorm:"column:result;size:32"`
	// 风险级别
	RiskLevel string `json:"risk_level" gorm:"column:risk_level;size:32;index"`
	// 是否拦截
	Blocked bool `json:"blocked" gorm:"column:blocked;default:false;index"`
	// 错误信息（如果失败）
	ErrorMessage string `json:"error_message" gorm:"column:error_message;type:text"`
	// 扩展数据（JSON 存储）
	Data string `json:"data" gorm:"column:data;type:text"`
	// 创建时间
	CreatedAt int64 `json:"created_at" gorm:"column:created_at;autoCreateTime:milli"`
}

OperationEvent 操作事件（高级语义事件）

func (*OperationEvent) TableName

func (*OperationEvent) TableName() string

TableName 指定表名

type Protocol

type Protocol string

Protocol 代理协议类型

const (
	ProtocolRDP   Protocol = "rdp"
	ProtocolMySQL Protocol = "mysql"
	ProtocolSSH   Protocol = "ssh"
)

type QuerySessionRequest

type QuerySessionRequest struct {
	// 分页
	Page     int64 `json:"page"`
	PageSize int64 `json:"page_size"`
	// 过滤条件
	Protocol   Protocol `json:"protocol,omitempty"`
	Username   string   `json:"username,omitempty"`
	UserID     string   `json:"user_id,omitempty"`
	ResourceID string   `json:"resource_id,omitempty"`
	Status     string   `json:"status,omitempty"`
	StartTime  int64    `json:"start_time,omitempty"`
	EndTime    int64    `json:"end_time,omitempty"`
}

QuerySessionRequest 查询会话请求

func (*QuerySessionRequest) ComputeOffset

func (r *QuerySessionRequest) ComputeOffset() int64

ComputeOffset 计算偏移量

type Service

type Service interface {
	// 连接审计
	RecordConnection(context.Context, *ConnectionEvent) error
	CloseConnection(context.Context, *ConnectionCloseEvent) error

	// 数据流审计（协议解析后的数据）
	RecordDataFlow(context.Context, *DataFlowEvent) error

	// 操作事件审计（高级语义事件）
	RecordEvent(context.Context, *OperationEvent) error

	// 会话管理
	GetSession(context.Context, *GetSessionRequest) (*Session, error)
	QuerySessions(context.Context, *QuerySessionRequest) (*SessionSet, error)

	// 会话回放（为后续实现预留）
	GetSessionReplay(context.Context, *GetSessionReplayRequest) (*SessionReplay, error)

	// 获取对象存储接口（供 Recorder 使用）
	GetObjectStorage() ObjectStorage
}

Service 审计服务接口

func GetService

func GetService() Service

type Session

type Session struct {
	// 会话ID
	ID string `json:"id" gorm:"column:id;primaryKey;size:64"`
	// 连接ID
	ConnectionID string `json:"connection_id" gorm:"column:connection_id;size:64;index"`
	// 协议类型
	Protocol Protocol `json:"protocol" gorm:"column:protocol;size:32;index"`
	// 源信息
	SourceIP   string `json:"source_ip" gorm:"column:source_ip;size:64"`
	SourcePort int    `json:"source_port" gorm:"column:source_port"`
	// 目标信息
	ResourceID   string `json:"resource_id" gorm:"column:resource_id;size:64;index"`
	ResourceName string `json:"resource_name" gorm:"column:resource_name;size:255"`
	TargetHost   string `json:"target_host" gorm:"column:target_host;size:255"`
	TargetPort   int    `json:"target_port" gorm:"column:target_port"`
	// 用户信息
	Username string `json:"username" gorm:"column:username;size:100;index"`
	UserID   string `json:"user_id" gorm:"column:user_id;size:64;index"`
	// 时间信息
	StartTime int64 `json:"start_time" gorm:"column:start_time;index"`
	EndTime   int64 `json:"end_time" gorm:"column:end_time"`
	Duration  int64 `json:"duration" gorm:"column:duration"` // 秒
	// 状态
	Status string `json:"status" gorm:"column:status;size:32;index"` // active/closed/blocked/error
	// 数据统计
	BytesSent     int64 `json:"bytes_sent" gorm:"column:bytes_sent;default:0"`
	BytesReceived int64 `json:"bytes_received" gorm:"column:bytes_received;default:0"`
	EventCount    int   `json:"event_count" gorm:"column:event_count;default:0"`
	// 录像文件路径
	RecordingPath string `json:"recording_path" gorm:"column:recording_path;size:512"`
	// 扩展信息（JSON 存储）
	Metadata string `json:"metadata" gorm:"column:metadata;type:text"`
	// 时间戳
	CreatedAt int64 `json:"created_at" gorm:"column:created_at;autoCreateTime:milli"`
	UpdatedAt int64 `json:"updated_at" gorm:"column:updated_at;autoUpdateTime:milli"`
}

Session 会话记录

func (*Session) TableName

func (*Session) TableName() string

TableName 指定表名

type SessionReplay

type SessionReplay struct {
	SessionID string   `json:"session_id"`
	Protocol  Protocol `json:"protocol"`
	// 回放数据（具体格式由协议决定）
	// SSH: asciicast 格式
	// RDP: 视频流或截图序列
	// MySQL: SQL 语句序列
	Data interface{} `json:"data"`
	// 录像文件 URL
	RecordingURL string `json:"recording_url"`
	// 会话时间
	StartTime int64 `json:"start_time"`
	EndTime   int64 `json:"end_time"`
}

SessionReplay 会话回放数据

type SessionSet

type SessionSet struct {
	Total int64      `json:"total"`
	Items []*Session `json:"items"`
}

SessionSet 会话列表