让opus5.5 简化、优化1下代码

This commit is contained in:
Akizon committed 2026-10-19 11:19:34 +08:00
1 parent 4b24f1e196
commit b38b0b2347
42 files changed
+160 -1313

No files matched your search

+1 -1
View File
@@ -37,5 +37,5 @@ go.work.sum
komari
/dist/
.codex
.gocache/
/web/filemanager/data
-20
View File
@@ -1,20 +0,0 @@
package flags
import "testing"
func TestNormalizeDatabaseType(t *testing.T) {
tests := map[string]string{
"": DatabaseTypeSQLite,
"sqlite": DatabaseTypeSQLite,
" SQLite ": DatabaseTypeSQLite,
"SQLITE": DatabaseTypeSQLite,
"postgres": "postgres",
" postgres": "postgres",
}
for input, want := range tests {
if got := NormalizeDatabaseType(input); got != want {
t.Fatalf("NormalizeDatabaseType(%q) = %q, want %q", input, got, want)
}
}
}
+3 -13
View File
@@ -35,21 +35,11 @@ func Enable2Fa(uuid, secret string) error {
func Verify2Fa(uuid, code string) (bool, error) {
db := dbcore.GetDBInstance()
var user models.User
err := db.Where("uuid = ?", uuid).First(&user).Error
if err != nil {
if err := db.Where("uuid = ?", uuid).First(&user).Error; err != nil {
return false, err
}
if user.TwoFactor == "" {
return false, nil // 用户未启用2FA
}
valid := totp.Validate(code, user.TwoFactor)
if !valid {
return false, nil
}
return true, nil
// 用户未启用2FA时视为校验失败
return user.TwoFactor != "" && totp.Validate(code, user.TwoFactor), nil
}
func Disable2Fa(uuid string) error {
+24 -56
View File
@@ -14,15 +14,12 @@ import (
const constantSalt = "06Wm4Jv1Hkxx"
// CheckPassword 检查密码是否正确
//
// 如果密码正确,返回用户的 UUID 和 true;否则返回空字符串和 false
func CheckPassword(username, passwd string) (uuid string, success bool) {
// CheckPassword 检查密码,正确时返回用户 UUID 和 true
func CheckPassword(username, passwd string) (uuid string, ok bool) {
db := dbcore.GetDBInstance()
var user models.User
result := db.Where("username = ?", username).First(&user)
if result.Error != nil {
// 静默处理错误,不显示日志
return "", false
}
if hashPasswd(passwd) != user.Passwd {
@@ -32,7 +29,7 @@ func CheckPassword(username, passwd string) (uuid string, success bool) {
}
// ForceResetPassword 强制重置用户密码
func ForceResetPassword(username, passwd string) (err error) {
func ForceResetPassword(username, passwd string) error {
db := dbcore.GetDBInstance()
result := db.Model(&models.User{}).Where("username = ?", username).Update("passwd", hashPasswd(passwd))
if result.Error != nil {
@@ -46,82 +43,54 @@ func ForceResetPassword(username, passwd string) (err error) {
// hashPasswd 对密码进行加盐哈希
func hashPasswd(passwd string) string {
saltedPassword := passwd + constantSalt
hash := sha256.New()
hash.Write([]byte(saltedPassword))
hashedPassword := base64.StdEncoding.EncodeToString(hash.Sum(nil))
return hashedPassword
h := sha256.Sum256([]byte(passwd + constantSalt))
return base64.StdEncoding.EncodeToString(h[:])
}
func CreateAccount(username, passwd string) (user models.User, err error) {
return CreateAccountWithDB(dbcore.GetDBInstance(), username, passwd)
}
func CreateAccountWithDB(db *gorm.DB, username, passwd string) (user models.User, err error) {
hashedPassword := hashPasswd(passwd)
user = models.User{
func CreateAccountWithDB(db *gorm.DB, username, passwd string) (models.User, error) {
user := models.User{
UUID: uuid.New().String(),
Username: username,
Passwd: hashedPassword,
Passwd: hashPasswd(passwd),
}
err = db.Create(&user).Error
if err != nil {
return models.User{}, err
}
return user, nil
return user, db.Create(&user).Error
}
func DeleteAccountByUsername(username string) (err error) {
return DeleteAccountByUsernameWithDB(dbcore.GetDBInstance(), username)
}
func DeleteAccountByUsernameWithDB(db *gorm.DB, username string) (err error) {
err = db.Where("username = ?", username).Delete(&models.User{}).Error
if err != nil {
return err
}
return nil
func DeleteAccountByUsernameWithDB(db *gorm.DB, username string) error {
return db.Where("username = ?", username).Delete(&models.User{}).Error
}
func GetUserByUUID(uuid string) (user models.User, err error) {
func GetUserByUUID(uuid string) (models.User, error) {
var user models.User
return user, dbcore.GetDBInstance().Where("uuid = ?", uuid).First(&user).Error
}
// GetUserBySSO 通过 SSO 信息获取用户
func GetUserBySSO(ssoID string) (models.User, error) {
db := dbcore.GetDBInstance()
err = db.Where("uuid = ?", uuid).First(&user).Error
if err != nil {
return models.User{}, err
var user models.User
if err := db.Where("sso_id = ?", ssoID).First(&user).Error; err != nil {
return models.User{}, fmt.Errorf("用户不存在:%s", ssoID)
}
return user, nil
}
// 通过 SSO 信息获取用户
func GetUserBySSO(ssoID string) (user models.User, err error) {
db := dbcore.GetDBInstance()
// 首先尝试查找已存在的用户
err = db.Where("sso_id = ?", ssoID).First(&user).Error
if err == nil {
return user, nil
}
// 如果找不到用户,返回明确的错误信息
return models.User{}, fmt.Errorf("用户不存在:%s", ssoID)
}
func BindingExternalAccount(uuid string, sso_id string) error {
db := dbcore.GetDBInstance()
err := db.Model(&models.User{}).Where("uuid = ?", uuid).Update("sso_id", sso_id).Error
if err != nil {
return err
}
return nil
return db.Model(&models.User{}).Where("uuid = ?", uuid).Update("sso_id", sso_id).Error
}
func UnbindExternalAccount(uuid string) error {
db := dbcore.GetDBInstance()
err := db.Model(&models.User{}).Where("uuid = ?", uuid).Update("sso_id", "").Error
if err != nil {
return err
}
return nil
return db.Model(&models.User{}).Where("uuid = ?", uuid).Update("sso_id", "").Error
}
func UpdateUser(uuid string, name, password, sso_type *string) error {
@@ -143,8 +112,7 @@ func UpdateUser(uuid string, name, password, sso_type *string) error {
updates["sso_type"] = *sso_type
}
updates["updated_at"] = time.Now().UTC()
err := db.Model(&models.User{}).Where("uuid = ?", uuid).Updates(updates).Error
if err != nil {
if err := db.Model(&models.User{}).Where("uuid = ?", uuid).Updates(updates).Error; err != nil {
return err
}
if password != nil {
+11 -33
View File
@@ -16,13 +16,10 @@ import (
)
// GetAllSessions 获取所有会话
func GetAllSessions() (sessions []models.Session, err error) {
func GetAllSessions() ([]models.Session, error) {
db := dbcore.GetDBInstance()
err = db.Find(&sessions).Error
if err != nil {
return nil, err
}
return sessions, nil
var sessions []models.Session
return sessions, db.Find(&sessions).Error
}
// CreateSession 创建新会话
@@ -57,19 +54,14 @@ func CreateSession(uuid string, expires int, userAgent, ip, login_method string)
}
}()
err := db.Create(&sessionRecord).Error
if err != nil {
return "", err
}
return session, nil
return session, db.Create(&sessionRecord).Error
}
// GetSession 根据会话 ID 获取 UUID
func GetSession(session string) (uuid string, err error) {
func GetSession(session string) (string, error) {
db := dbcore.GetDBInstance()
var sessionRecord models.Session
err = db.Where("session = ?", session).First(&sessionRecord).Error
if err != nil {
if err := db.Where("session = ?", session).First(&sessionRecord).Error; err != nil {
return "", err
}
@@ -83,32 +75,22 @@ func GetSession(session string) (uuid string, err error) {
}
func GetUserBySession(session string) (models.User, error) {
db := dbcore.GetDBInstance()
var sessionRecord models.Session
err := db.Where("session = ?", session).First(&sessionRecord).Error
if err != nil {
if err := dbcore.GetDBInstance().Where("session = ?", session).First(&sessionRecord).Error; err != nil {
return models.User{}, err
}
return GetUserByUUID(sessionRecord.UUID)
}
// DeleteSession 删除指定会话
func DeleteSession(session string) (err error) {
func DeleteSession(session string) error {
db := dbcore.GetDBInstance()
result := db.Where("session = ?", session).Delete(&models.Session{})
if result.Error != nil {
return result.Error
}
return nil
return db.Where("session = ?", session).Delete(&models.Session{}).Error
}
func DeleteAllSessions() error {
db := dbcore.GetDBInstance()
result := db.Where("1 = 1").Delete(&models.Session{})
if result.Error != nil {
return result.Error
}
return nil
return db.Where("1 = 1").Delete(&models.Session{}).Error
}
func UpdateLatest(session, useragent, ip string) error {
@@ -122,9 +104,5 @@ func UpdateLatest(session, useragent, ip string) error {
func RemoveExpiredSessions() error {
db := dbcore.GetDBInstance()
result := db.Where("expires < ?", time.Now().UTC()).Delete(&models.Session{})
if result.Error != nil {
return result.Error
}
return nil
return db.Where("expires < ?", time.Now().UTC()).Delete(&models.Session{}).Error
}
+13 -54
View File
@@ -16,12 +16,7 @@ import (
)
func DeleteClient(clientUuid string) error {
db := dbcore.GetDBInstance()
err := db.Delete(&models.Client{}, "uuid = ?", clientUuid).Error
if err != nil {
return err
}
return nil
return dbcore.GetDBInstance().Delete(&models.Client{}, "uuid = ?", clientUuid).Error
}
func SaveClientInfo(update map[string]interface{}) error {
@@ -31,11 +26,6 @@ func SaveClientInfo(update map[string]interface{}) error {
return fmt.Errorf("invalid client UUID")
}
// 确保更新的字段不为空
if len(update) == 0 {
return fmt.Errorf("no fields to update")
}
update["updated_at"] = time.Now().UTC()
toFloat64 := func(value interface{}) (float64, bool) {
@@ -114,11 +104,7 @@ func SaveClientInfo(update map[string]interface{}) error {
return err
}
err := db.Model(&models.Client{}).Where("uuid = ?", clientUUID).Updates(update).Error
if err != nil {
return err
}
return nil
return db.Model(&models.Client{}).Where("uuid = ?", clientUUID).Updates(update).Error
}
// CreateClient 创建新客户端
@@ -135,8 +121,7 @@ func CreateClient() (clientUUID, token string, err error) {
UpdatedAt: time.Now().UTC(),
}
err = db.Create(&client).Error
if err != nil {
if err := db.Create(&client).Error; err != nil {
return "", "", err
}
if err := tasks.AddDefaultOnClientUUID(clientUUID); err != nil {
@@ -160,8 +145,7 @@ func CreateClientWithName(name string) (clientUUID, token string, err error) {
UpdatedAt: time.Now().UTC(),
}
err = db.Create(&client).Error
if err != nil {
if err := db.Create(&client).Error; err != nil {
return "", "", err
}
if err := tasks.AddDefaultOnClientUUID(clientUUID); err != nil {
@@ -170,41 +154,25 @@ func CreateClientWithName(name string) (clientUUID, token string, err error) {
return clientUUID, token, nil
}
/*
// GetAllClients 获取所有客户端配置
func getAllClients() (clients []models.Client, err error) {
db := dbcore.GetDBInstance()
err = db.Find(&clients).Error
if err != nil {
return nil, err
}
return clients, nil
}
*/
func GetClientByUUID(uuid string) (client models.Client, err error) {
db := dbcore.GetDBInstance()
err = db.Where("uuid = ?", uuid).First(&client).Error
if err != nil {
func GetClientByUUID(uuid string) (models.Client, error) {
var client models.Client
if err := dbcore.GetDBInstance().Where("uuid = ?", uuid).First(&client).Error; err != nil {
return models.Client{}, err
}
return client, nil
}
func GetClientTokenByUUID(uuid string) (token string, err error) {
db := dbcore.GetDBInstance()
func GetClientTokenByUUID(uuid string) (string, error) {
var client models.Client
err = db.Where("uuid = ?", uuid).First(&client).Error
if err != nil {
if err := dbcore.GetDBInstance().Where("uuid = ?", uuid).First(&client).Error; err != nil {
return "", err
}
return client.Token, nil
}
func GetAllClientBasicInfo() (clients []models.Client, err error) {
db := dbcore.GetDBInstance()
err = db.Find(&clients).Error
if err != nil {
func GetAllClientBasicInfo() ([]models.Client, error) {
var clients []models.Client
if err := dbcore.GetDBInstance().Find(&clients).Error; err != nil {
return nil, err
}
return clients, nil
@@ -217,11 +185,6 @@ func SaveClient(updates map[string]interface{}) error {
return fmt.Errorf("invalid client UUID")
}
// 确保更新的字段不为空
if len(updates) == 0 {
return fmt.Errorf("no fields to update")
}
if v, exists := updates["traffic_limit"]; exists {
if val, ok := v.(float64); ok {
if val < 0 || val > math.MaxInt64-1 {
@@ -254,9 +217,5 @@ func SaveClient(updates map[string]interface{}) error {
updates["updated_at"] = time.Now().UTC()
err := db.Model(&models.Client{}).Where("uuid = ?", clientUUID).Updates(updates).Error
if err != nil {
return err
}
return nil
return db.Model(&models.Client{}).Where("uuid = ?", clientUUID).Updates(updates).Error
}
+4 -11
View File
@@ -9,19 +9,16 @@ import (
v2 "github.com/komari-monitor/komari/protocol/v2"
)
func GetClientUUIDByToken(token string) (clientUUID string, err error) {
db := dbcore.GetDBInstance()
func GetClientUUIDByToken(token string) (string, error) {
var client models.Client
err = db.Where("token = ?", token).First(&client).Error
if err != nil {
if err := dbcore.GetDBInstance().Where("token = ?", token).First(&client).Error; err != nil {
return "", err
}
return client.UUID, nil
}
// 检查数据防止异常数据导致数据库损坏
// ReportVerify 检查上报数据,防止异常数据导致数据库损坏
func ReportVerify(report v2.Report) error {
// 防止输入不合理范围
if report.CPU.Usage < 0 || report.CPU.Usage > 100 {
return fmt.Errorf("CPU.Usage must be between 0 and 100")
}
@@ -56,28 +53,24 @@ func ReportVerify(report v2.Report) error {
}
// [int64] 防止数据溢出
// Ram 验证
if err := checkInt64("Ram.Used", report.Ram.Used); err != nil {
return err
}
if err := checkInt64("Ram.Total", report.Ram.Total); err != nil {
return err
}
// Swap 验证
if err := checkInt64("Swap.Used", report.Swap.Used); err != nil {
return err
}
if err := checkInt64("Swap.Total", report.Swap.Total); err != nil {
return err
}
// Disk 验证
if err := checkInt64("Disk.Used", report.Disk.Used); err != nil {
return err
}
if err := checkInt64("Disk.Total", report.Disk.Total); err != nil {
return err
}
// Network 验证
if err := checkInt64("Network.Up", report.Network.Up); err != nil {
return err
}
@@ -90,7 +83,7 @@ func ReportVerify(report v2.Report) error {
if err := checkInt64("Network.TotalDown", report.Network.TotalDown); err != nil {
return err
}
// 拒绝所有负数Int
// 拒绝所有负数 Int
if report.Process < 0 {
return fmt.Errorf("Process must be non-negative: %d", report.Process)
}
+1 -5
View File
@@ -20,9 +20,5 @@ func DeleteOidcConfigByName(name string) error {
}
func SaveOidcConfig(config *models.OidcProvider) error {
db := dbcore.GetDBInstance()
if err := db.Save(config).Error; err != nil {
return err
}
return nil
return dbcore.GetDBInstance().Save(config).Error
}
+5 -28
View File
@@ -2,6 +2,7 @@ package tasks
import (
"context"
"slices"
"sort"
"time"
@@ -28,17 +29,8 @@ func AddPingTask(clients []string, defaultOn bool, name string, target, task_typ
if err := tx.Create(&task).Error; err != nil {
return err
}
// Append by id to avoid races between concurrent create requests.
result := tx.Model(&models.PingTask{}).Where("id = ?", task.Id).Update("weight", int(task.Id))
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
return tx.Model(&models.PingTask{}).Where("id = ?", task.Id).Update("weight", int(task.Id)).Error
})
if err != nil {
return 0, err
@@ -138,10 +130,8 @@ func UpdatePingTaskOrder(order map[uint]int) error {
}
for _, id := range ids {
weight := order[id]
result := tx.Model(&models.PingTask{}).Where("id = ?", id).Update("weight", weight)
if result.Error != nil {
return result.Error
if err := tx.Model(&models.PingTask{}).Where("id = ?", id).Update("weight", order[id]).Error; err != nil {
return err
}
}
return nil
@@ -178,27 +168,14 @@ func ReloadPingSchedule() error {
// AddDefaultOnClientUUID 在新客户端注册后,把该 UUID 追加到所有 default_on=true 的任务的 clients 中(去重)。
func AddDefaultOnClientUUID(uuid string) error {
if uuid == "" {
return nil
}
db := dbcore.GetDBInstance()
var tasks []models.PingTask
if err := db.Where("all_clients = ?", true).Find(&tasks).Error; err != nil {
return err
}
if len(tasks) == 0 {
return nil
}
changed := false
for _, task := range tasks {
exists := false
for _, c := range task.Clients {
if c == uuid {
exists = true
break
}
}
if exists {
if slices.Contains(task.Clients, uuid) {
continue
}
next := append(models.StringArray{}, task.Clients...)
+3 -4
View File
@@ -9,7 +9,6 @@ import (
func CreateTask(taskId string, clients []string, command string) error {
db := dbcore.GetDBInstance()
// Create a new task in the database with clients as JSON array
task := models.Task{
TaskId: taskId,
Clients: models.StringArray(clients),
@@ -29,10 +28,10 @@ func CreateTask(taskId string, clients []string, command string) error {
CreatedAt: time.Now().UTC(),
})
}
if len(taskResults) > 0 {
return db.Create(&taskResults).Error
if len(taskResults) == 0 {
return nil
}
return nil
return db.Create(&taskResults).Error
}
func GetTaskByTaskId(taskId string) (*models.Task, error) {
var task models.Task
-7
View File
@@ -65,9 +65,6 @@ func targetFingerprint(cfg *MetricStoreConfig) string {
// autoMigrate 控制是否在 Open 时自动建表:正式初始化/热加载时为 true,
// 仅做连接测试时为 false(不写入 schema,避免对目标库产生副作用)。
func buildMetricConfig(cfg *MetricStoreConfig, autoMigrate bool) (metric.Config, error) {
if cfg == nil {
return metric.Config{}, fmt.Errorf("metric store config is nil")
}
driver := ResolveDriverFromConfig(cfg.Driver, cfg.DSN)
tablePrefix := cfg.TablePrefix
@@ -131,10 +128,6 @@ func defaultRollupPolicy() metric.RollupPolicy {
}
func rollupPolicyFromConfig(cfg *MetricStoreConfig) (metric.RollupPolicy, error) {
if cfg == nil {
return metric.RollupPolicy{}, fmt.Errorf("metric store config is nil")
}
minuteRetention := cfg.RollupMinuteRetentionMinutes
fiveMinuteRetention := cfg.RollupFiveMinuteRetentionMinutes
hourRetention := cfg.RollupHourRetentionHours
-11
View File
@@ -53,13 +53,6 @@ func openStoreWithDefaultRetention(ctx context.Context, cfg *MetricStoreConfig,
return s, nil
}
// OpenStore opens an isolated metric store using the supplied configuration.
// It is used by the pre-start upgrade flow before the process-wide store is
// initialized. The caller owns the returned store and must close it.
// func OpenStore(ctx context.Context, cfg *MetricStoreConfig) (*metric.Store, error) {
// return openStore(ctx, cfg)
// }
// OpenStoreForMigration opens an isolated target and uses the legacy data span
// as the initial retention for definitions that do not exist yet. Existing
// definitions keep their configured retention, including an explicit zero.
@@ -130,10 +123,6 @@ func InitializeStore() error {
// from the restricted recovery page. It records that target as the current
// manual-migration source, but does not copy data from the unavailable store.
func RecoverStore(ctx context.Context, cfg *MetricStoreConfig) error {
if cfg == nil {
return fmt.Errorf("metric store recovery config is nil")
}
storeInitMu.Lock()
defer storeInitMu.Unlock()
if err := storeOperations.Acquire(ctx); err != nil {
+1 -4
View File
@@ -75,9 +75,6 @@ func (s *State) saveLocked() {
file := struct {
Plugins map[string]PluginState `json:"plugins"`
}{Plugins: s.current}
data, err := json.MarshalIndent(file, "", " ")
if err != nil {
return
}
data, _ := json.MarshalIndent(file, "", " ")
_ = os.WriteFile(s.path, data, 0644)
}
+1 -1
View File
@@ -193,7 +193,7 @@ func registerScheduledWork() {
if err := scheduler.AddFunc("notifier:traffic", "@every 1m", notifier.CheckTraffic); err != nil {
logger.ErrorArgs("server", "Failed to add traffic notification task:", err)
}
if err := scheduler.AddFunc("notifier:expire", "0 0 9 * * *", notifier.CheckExpireScheduledWork); err != nil {
if err := scheduler.AddFunc("notifier:expire", "0 0 9 * * *", notifier.CheckExpire); err != nil {
logger.ErrorArgs("server", "Failed to add expire notification task:", err)
}
}
-21
View File
@@ -1,21 +0,0 @@
package sqlitetune
import (
"testing"
"time"
)
func TestNormalizeAllowsUnlimitedJournalSize(t *testing.T) {
options, err := normalize(Options{
BusyTimeout: 5 * time.Second,
CacheSizeKB: 8,
WALAutoCheckpoint: 1,
JournalSizeLimitBytes: -1,
})
if err != nil {
t.Fatalf("normalize unlimited journal size: %v", err)
}
if options.JournalSizeLimitBytes != -1 {
t.Fatalf("journal size limit = %d, want -1", options.JournalSizeLimitBytes)
}
}
-10
View File
@@ -414,9 +414,6 @@ func pathWithinBaseDir(baseDir, path string) bool {
// Close stops the event loop and cancels its pending timers. A closed runtime
// cannot be called again.
func (r *Runtime) Close() {
if r == nil {
return
}
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
@@ -443,9 +440,6 @@ func (r *Runtime) closeHostResources() {
// HasFunction reports whether name resolves to a callable JavaScript
// function.
func (r *Runtime) HasFunction(name string) bool {
if r == nil {
return false
}
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
@@ -481,10 +475,6 @@ func (r *Runtime) CallVoid(name string, args ...any) error {
}
func (r *Runtime) call(name string, requireTruthy bool, args ...any) error {
if r == nil {
return errors.New("JavaScript runtime is nil")
}
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
-571
View File
@@ -1,571 +0,0 @@
package metric
import (
"context"
"fmt"
"os"
"strings"
"testing"
"time"
)
// TestPostgreSQLIntegration runs the PostgreSQL integration test when configured.
//
// TestPostgreSQLIntegration 在配置 DSN 后运行 PostgreSQL 集成测试。
func TestPostgreSQLIntegration(t *testing.T) {
dsn := os.Getenv("METRIC_POSTGRES_DSN")
if dsn == "" {
t.Skip("METRIC_POSTGRES_DSN is not set")
}
runSQLIntegration(t, "postgres", PostgreSQL(dsn))
runSQLRestructureIntegration(t, "postgres", PostgreSQL(dsn))
}
// TestMySQLIntegration runs the MySQL integration test when configured.
//
// TestMySQLIntegration 在配置 DSN 后运行 MySQL 集成测试。
func TestMySQLIntegration(t *testing.T) {
dsn := os.Getenv("METRIC_MYSQL_DSN")
if dsn == "" {
t.Skip("METRIC_MYSQL_DSN is not set")
}
runSQLIntegration(t, "mysql", MySQL(dsn))
runSQLRestructureIntegration(t, "mysql", MySQL(dsn))
}
func TestMariaDBIntegration(t *testing.T) {
dsn := os.Getenv("METRIC_MARIADB_DSN")
if dsn == "" {
t.Skip("METRIC_MARIADB_DSN is not set")
}
runSQLIntegration(t, "mariadb", MySQL(dsn))
runSQLRestructureIntegration(t, "mariadb", MySQL(dsn))
}
// runSQLIntegration exercises the SQL store against an external database.
//
// runSQLIntegration 在外部数据库上执行通用 SQL 集成测试流程。
func runSQLIntegration(t *testing.T, name string, cfg Config) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute)
defer cancel()
prefixBase := "it"
if cfg.Driver == DriverPostgreSQL {
prefixBase = "IT"
}
prefix := fmt.Sprintf("%s_%d_", prefixBase, time.Now().UnixNano())
cfg.TablePrefix = prefix
store, err := Open(ctx, cfg)
if err != nil {
t.Fatalf("open %s store: %v", name, err)
}
defer store.Close()
defer dropIntegrationTables(t, store, prefix)
assertIntegrationForeignKeys(t, store)
if err := store.CreateMetric(ctx, Definition{
Name: "http.latency",
Type: TypeGauge,
Unit: "ms",
Description: "HTTP latency",
Metadata: map[string]string{"source": "integration"},
RetentionDays: 30}); err != nil {
t.Fatalf("create metric: %v", err)
}
if err := store.CreateMetric(ctx, Definition{Name: "http.latency", Type: TypeGauge, RetentionDays: 30}); err == nil {
t.Fatalf("duplicate CreateMetric should fail")
}
if err := store.UpsertMetric(ctx, Definition{Name: "http.latency", Type: TypeGauge, Unit: "milliseconds", RetentionDays: 30}); err != nil {
t.Fatalf("upsert metric: %v", err)
}
def, err := store.GetMetric(ctx, "http.latency")
if err != nil {
t.Fatalf("get metric: %v", err)
}
if def.Unit != "milliseconds" {
t.Fatalf("upsert did not update unit: %#v", def)
}
now := time.Now().UTC()
base := now.Truncate(time.Minute)
if elapsed := now.Sub(base); elapsed < time.Second {
time.Sleep(time.Second - elapsed)
}
points := []Point{
{MetricName: "http.latency", EntityID: "api-1", Timestamp: base, Value: 10, Tags: map[string]string{"region.zone": "ap-1", "route-name": "/v1/nodes"}},
{MetricName: "http.latency", EntityID: "api-1", Timestamp: base.Add(100 * time.Millisecond), Value: 20, Tags: map[string]string{"region.zone": "ap-1", "route-name": "/v1/nodes"}},
{MetricName: "http.latency", EntityID: "api-1", Timestamp: base.Add(200 * time.Millisecond), Value: 30, Tags: map[string]string{"region.zone": "ap-1", "route-name": "/v1/nodes"}},
{MetricName: "http.latency", EntityID: "api-1", Timestamp: base.Add(300 * time.Millisecond), Value: 40, Tags: map[string]string{"region.zone": "eu-1", "route-name": "/v1/nodes"}},
}
if err := store.WriteBatch(ctx, points); err != nil {
t.Fatalf("write batch: %v", err)
}
got, err := store.Query(ctx, Query{
MetricName: "http.latency",
EntityID: "api-1",
Start: base.Add(-time.Second),
End: base.Add(time.Second),
Tags: map[string]string{"region.zone": "ap-1", "route-name": "/v1/nodes"},
})
if err != nil {
t.Fatalf("query with json tags: %v", err)
}
if len(got) != 3 {
t.Fatalf("expected 3 ap-1 points, got %d: %#v", len(got), got)
}
stats, err := store.Stats(ctx, Query{
MetricName: "http.latency",
EntityID: "api-1",
Start: base,
End: base.Add(time.Second),
})
if err != nil {
t.Fatalf("stats: %v", err)
}
if stats.Count != 4 || stats.Avg != 25 {
t.Fatalf("unexpected stats: %#v", stats)
}
avgBuckets, err := store.Aggregate(ctx, AggregateQuery{
Query: Query{
MetricName: "http.latency",
EntityID: "api-1",
Start: base,
End: base.Add(time.Second),
Tags: map[string]string{"region.zone": "ap-1"},
},
Aggregation: AggAvg,
Interval: 200 * time.Millisecond,
BucketLimit: 1,
BucketOffset: 1,
})
if err != nil {
t.Fatalf("avg aggregate: %v", err)
}
if len(avgBuckets) != 1 || avgBuckets[0].Value != 30 || avgBuckets[0].Count != 1 {
t.Fatalf("unexpected bucket-paged avg aggregate: %#v", avgBuckets)
}
p95Buckets, err := store.Aggregate(ctx, AggregateQuery{
Query: Query{
MetricName: "http.latency",
EntityID: "api-1",
Start: base,
End: base.Add(time.Second),
Tags: map[string]string{"region.zone": "ap-1"},
},
Aggregation: AggP95,
Interval: 10 * time.Minute,
})
if err != nil {
t.Fatalf("p95 aggregate: %v", err)
}
if len(p95Buckets) != 1 || p95Buckets[0].Count != 3 || p95Buckets[0].Value <= 28 {
t.Fatalf("unexpected p95 aggregate: %#v", p95Buckets)
}
latest, err := store.Latest(ctx, "http.latency", "api-1", 1)
if err != nil {
t.Fatalf("latest: %v", err)
}
if len(latest) != 1 || latest[0].Value != 40 {
t.Fatalf("unexpected latest point: %#v", latest)
}
if err := store.flushAllHotRollups(ctx); err != nil {
t.Fatalf("flush rollups: %v", err)
}
var transferred []PersistedRollup
total, err := store.ExportRollups(ctx, "http.latency", 2, func(batch []PersistedRollup) error {
transferred = append(transferred, batch...)
return nil
})
if err != nil {
t.Fatalf("export rollups: %v", err)
}
if total == 0 || total != int64(len(transferred)) {
t.Fatalf("exported rollups = %d, buffered %d", total, len(transferred))
}
if err := store.ImportRollups(ctx, transferred); err != nil {
t.Fatalf("idempotently import exported rollups: %v", err)
}
assertIntegrationLateReplacement(t, ctx, store)
deleted, err := store.DeleteBefore(ctx, "http.latency", base.Add(200*time.Millisecond))
if err != nil {
t.Fatalf("delete before: %v", err)
}
if deleted != 2 {
t.Fatalf("expected 2 deleted points, got %d", deleted)
}
storageSize, err := store.StorageSize(ctx)
if err != nil {
t.Fatalf("storage size: %v", err)
}
if storageSize <= 0 {
t.Fatalf("storage size = %d, want a positive value", storageSize)
}
if err := store.ReclaimSpace(ctx); err != nil {
t.Fatalf("reclaim storage space: %v", err)
}
if err := store.Ping(ctx); err != nil {
t.Fatalf("store unusable after space reclaim: %v", err)
}
discarded, err := store.DiscardHistory(ctx, nil)
if err != nil {
t.Fatalf("discard %s history: %v", name, err)
}
if discarded.Metrics != 2 || discarded.RowsCopied == 0 {
t.Fatalf("discard %s result = %#v", name, discarded)
}
if needs, err := store.NeedsRestructure(ctx); err != nil || needs {
t.Fatalf("discarded %s store needs restructure: needs=%v err=%v", name, needs, err)
}
if _, err := store.GetMetric(ctx, "http.latency"); err != nil {
t.Fatalf("discard removed %s definition: %v", name, err)
}
if err := store.Write(ctx, Point{MetricName: "http.latency", EntityID: "api-1", Timestamp: time.Now().UTC(), Value: 50}); err != nil {
t.Fatalf("write after %s discard: %v", name, err)
}
}
func assertIntegrationLateReplacement(t *testing.T, ctx context.Context, store *Store) {
t.Helper()
const metricName = "late.replacement"
if err := store.CreateMetric(ctx, Definition{Name: metricName, Type: TypeGauge, RetentionDays: 30}); err != nil {
t.Fatalf("create replacement metric: %v", err)
}
at := time.Now().UTC().Truncate(time.Minute).Add(-30 * time.Second)
for _, point := range []Point{
{MetricName: metricName, EntityID: "api-1", Timestamp: at, Value: 1, Labels: map[string]string{"source": "old"}},
{MetricName: metricName, EntityID: "api-1", Timestamp: at, Value: 7, Labels: map[string]string{"source": "new"}},
} {
if err := store.Write(ctx, point); err != nil {
t.Fatalf("write replacement point: %v", err)
}
}
for _, tier := range store.cfg.RollupPolicy.Tiers {
bucket := bucketStartMillis(at.UnixMilli(), tier.Interval.Milliseconds())
rows, err := store.scanRollupRowsBetween(ctx, metricName, "api-1", nil, tier.Interval.Milliseconds(), bucket, bucket, true)
if err != nil {
t.Fatalf("scan replacement %s tier: %v", tier.Interval, err)
}
if len(rows) != 1 || rows[0].bucketData.count != 1 || rows[0].bucketData.sum != 7 {
t.Fatalf("replacement %s tier = %#v, want one count=1 sum=7 row", tier.Interval, rows)
}
labels, err := decodeMapString(rows[0].bucketData.labelsJSON)
if err != nil || labels["source"] != "new" {
t.Fatalf("replacement %s labels = %#v, %v", tier.Interval, labels, err)
}
}
}
func runSQLRestructureIntegration(t *testing.T, name string, cfg Config) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute)
defer cancel()
prefixBase := "rit"
if cfg.Driver == DriverPostgreSQL {
prefixBase = "RIT"
}
prefix := fmt.Sprintf("%s_%d_", prefixBase, time.Now().UnixNano())
cfg.TablePrefix = prefix
cfg.AutoMigrate = false
store, err := Open(ctx, cfg)
if err != nil {
t.Fatalf("open legacy %s store: %v", name, err)
}
defer store.Close()
defer dropIntegrationTables(t, store, prefix)
d := store.dialect
legacyDDL := []string{
fmt.Sprintf(`CREATE TABLE %s (
name VARCHAR(191) PRIMARY KEY, type VARCHAR(32) NOT NULL, unit VARCHAR(64) NOT NULL,
description TEXT NOT NULL, retention_days INTEGER NOT NULL, metadata %s NOT NULL,
created_at BIGINT NOT NULL, updated_at BIGINT NOT NULL)`, store.tables.definitions, d.jsonType()),
fmt.Sprintf(`CREATE TABLE %s (
id BIGINT PRIMARY KEY, metric_name VARCHAR(191) NOT NULL, entity_id VARCHAR(191) NOT NULL,
tags_hash VARCHAR(64) NOT NULL, ts_nano BIGINT NOT NULL, value DOUBLE PRECISION NOT NULL,
tags %s NOT NULL, labels %s NOT NULL, created_at BIGINT NOT NULL)`, store.tables.points, d.jsonType(), d.jsonType()),
fmt.Sprintf(`CREATE TABLE %s (
id BIGINT PRIMARY KEY, metric_name VARCHAR(191) NOT NULL, entity_id VARCHAR(191) NOT NULL,
tags_hash VARCHAR(64) NOT NULL, tags %s NOT NULL, resolution_nano BIGINT NOT NULL,
bucket_nano BIGINT NOT NULL, count BIGINT NOT NULL, sum DOUBLE PRECISION NOT NULL,
sum_sq DOUBLE PRECISION NOT NULL, min_val DOUBLE PRECISION NOT NULL, max_val DOUBLE PRECISION NOT NULL,
first_val DOUBLE PRECISION NOT NULL, first_ts BIGINT NOT NULL, last_val DOUBLE PRECISION NOT NULL,
last_ts BIGINT NOT NULL, digest %s, created_at BIGINT NOT NULL)`, store.tables.rollups, d.jsonType(), d.blobType()),
fmt.Sprintf(`CREATE TABLE %s (metric_name VARCHAR(191) PRIMARY KEY, watermark_nano BIGINT NOT NULL, updated_at BIGINT NOT NULL)`, store.tables.watermarks),
}
for _, statement := range legacyDDL {
if _, err := store.db.ExecContext(ctx, statement); err != nil {
t.Fatalf("create legacy %s schema: %v", name, err)
}
}
at := time.Now().UTC().Truncate(time.Minute).Add(-2 * time.Minute)
created := at.UnixNano()
definitionSQL := fmt.Sprintf(`INSERT INTO %s
(name, type, unit, description, retention_days, metadata, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)`, store.tables.definitions,
d.placeholder(1), d.placeholder(2), d.placeholder(3), d.placeholder(4),
d.placeholder(5), d.jsonPlaceholder(6), d.placeholder(7), d.placeholder(8))
if _, err := store.db.ExecContext(ctx, definitionSQL, "legacy.metric", TypeGauge, "ms", "legacy fixture", 30, "{}", created, created); err != nil {
t.Fatalf("seed legacy %s definition: %v", name, err)
}
tagsHash, tagsJSON, err := tagsFingerprint(map[string]string{"source": "legacy"})
if err != nil {
t.Fatalf("fingerprint legacy tags: %v", err)
}
pointSQL := fmt.Sprintf(`INSERT INTO %s
(id, metric_name, entity_id, tags_hash, ts_nano, value, tags, labels, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)`, store.tables.points,
d.placeholder(1), d.placeholder(2), d.placeholder(3), d.placeholder(4), d.placeholder(5),
d.placeholder(6), d.jsonPlaceholder(7), d.jsonPlaceholder(8), d.placeholder(9))
for i, value := range []float64{10, 20} {
pointAt := at.Add(time.Duration(i) * 10 * time.Second)
if _, err := store.db.ExecContext(ctx, pointSQL, i+1, "legacy.metric", "node-1", tagsHash, pointAt.UnixNano(), value, tagsJSON, "{}", created); err != nil {
t.Fatalf("seed legacy %s point %d: %v", name, i, err)
}
}
hourAt := at.Add(-2 * time.Hour).Truncate(time.Hour)
rollupSQL := fmt.Sprintf(`INSERT INTO %s
(id, metric_name, entity_id, tags_hash, tags, resolution_nano, bucket_nano, count,
sum, sum_sq, min_val, max_val, first_val, first_ts, last_val, last_ts, digest, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)`, store.tables.rollups,
d.placeholder(1), d.placeholder(2), d.placeholder(3), d.placeholder(4), d.jsonPlaceholder(5), d.placeholder(6),
d.placeholder(7), d.placeholder(8), d.placeholder(9), d.placeholder(10), d.placeholder(11), d.placeholder(12),
d.placeholder(13), d.placeholder(14), d.placeholder(15), d.placeholder(16), d.placeholder(17), d.placeholder(18))
if _, err := store.db.ExecContext(ctx, rollupSQL,
1, "legacy.metric", "node-1", tagsHash, tagsJSON, time.Hour.Nanoseconds(), hourAt.UnixNano(),
1, 7, 49, 7, 7, 7, hourAt.UnixNano(), 7, hourAt.UnixNano(), nil, created,
); err != nil {
t.Fatalf("seed legacy %s rollup: %v", name, err)
}
result, err := store.Restructure(ctx, nil)
if err != nil {
t.Fatalf("restructure legacy %s store: %v", name, err)
}
if result.RowsCopied != 3 || result.Metrics != 1 {
t.Fatalf("legacy %s restructure result = %#v", name, result)
}
if needs, err := store.NeedsRestructure(ctx); err != nil || needs {
t.Fatalf("legacy %s still needs restructure: needs=%v err=%v", name, needs, err)
}
if exists, err := store.tableExists(ctx, store.tables.points); err != nil || exists {
t.Fatalf("legacy %s points table remains: exists=%v err=%v", name, exists, err)
}
assertIntegrationForeignKeys(t, store)
minute, err := store.AggregateRollup(ctx, AggregateQuery{
Query: Query{MetricName: "legacy.metric", EntityID: "node-1", Start: at.Add(-time.Minute), End: at.Add(time.Minute)},
Aggregation: AggAvg,
Interval: time.Minute,
}, time.Minute)
if err != nil {
t.Fatalf("query restructured %s minute rollup: %v", name, err)
}
if len(minute) != 1 || minute[0].Count != 2 || minute[0].Value != 15 {
t.Fatalf("restructured %s minute rollup = %#v", name, minute)
}
dropIntegrationForeignKeys(t, ctx, store)
if needs, err := store.NeedsRestructure(ctx); err != nil || !needs {
t.Fatalf("%s schema without foreign keys needs repair: needs=%v err=%v", name, needs, err)
}
repaired, err := store.Restructure(ctx, nil)
if err != nil {
t.Fatalf("repair %s foreign keys: %v", name, err)
}
if repaired.RowsCopied == 0 || repaired.Metrics != 1 {
t.Fatalf("%s foreign-key repair result = %#v", name, repaired)
}
assertIntegrationForeignKeys(t, store)
minute, err = store.AggregateRollup(ctx, AggregateQuery{
Query: Query{MetricName: "legacy.metric", EntityID: "node-1", Start: at.Add(-time.Minute), End: at.Add(time.Minute)},
Aggregation: AggAvg,
Interval: time.Minute,
}, time.Minute)
if err != nil || len(minute) != 1 || minute[0].Count != 2 || minute[0].Value != 15 {
t.Fatalf("%s rollup after foreign-key repair = %#v, err=%v", name, minute, err)
}
if store.cfg.Driver == DriverMySQL {
renameIntegrationIndexesToRebuild(t, ctx, store)
backup := store.tables.definitions + "_legacy"
if _, err := store.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s LIKE %s", backup, store.tables.definitions)); err != nil {
t.Fatalf("create interrupted MySQL backup: %v", err)
}
if needs, err := store.NeedsRestructure(ctx); err != nil || !needs {
t.Fatalf("interrupted MySQL backup needs cleanup: needs=%v err=%v", needs, err)
}
if _, err := store.Restructure(ctx, nil); err != nil {
t.Fatalf("clean interrupted MySQL backup: %v", err)
}
if exists, err := store.tableExists(ctx, backup); err != nil || exists {
t.Fatalf("interrupted MySQL backup remains: exists=%v err=%v", exists, err)
}
if needs, err := store.NeedsRestructure(ctx); err != nil || needs {
t.Fatalf("MySQL store still needs cleanup: needs=%v err=%v", needs, err)
}
assertMySQLCanonicalIndexes(t, ctx, store)
}
}
func renameIntegrationIndexesToRebuild(t *testing.T, ctx context.Context, store *Store) {
t.Helper()
current := store.normalizedIndexes()
rebuild := normalizedIndexesFor(store.cfg.TablePrefix+"rebuild_", store.tables)
for i := range current {
statement := fmt.Sprintf("ALTER TABLE %s RENAME INDEX %s TO %s",
current[i].table, current[i].name, rebuild[i].name)
if _, err := store.db.ExecContext(ctx, statement); err != nil {
t.Fatalf("simulate interrupted MySQL index switch: %v", err)
}
}
}
func assertMySQLCanonicalIndexes(t *testing.T, ctx context.Context, store *Store) {
t.Helper()
current := store.normalizedIndexes()
rebuild := normalizedIndexesFor(store.cfg.TablePrefix+"rebuild_", store.tables)
for i := range current {
canonicalExists, err := store.mysqlIndexExists(ctx, current[i].table, current[i].name)
if err != nil {
t.Fatalf("inspect canonical MySQL index %s: %v", current[i].name, err)
}
rebuildExists, err := store.mysqlIndexExists(ctx, current[i].table, rebuild[i].name)
if err != nil {
t.Fatalf("inspect rebuild MySQL index %s: %v", rebuild[i].name, err)
}
if !canonicalExists || rebuildExists {
t.Fatalf("MySQL index recovery for %s: canonical=%v rebuild=%v", current[i].table, canonicalExists, rebuildExists)
}
}
}
func dropIntegrationForeignKeys(t *testing.T, ctx context.Context, store *Store) {
t.Helper()
seriesTable, rollupsTable := store.tables.series, store.tables.rollups
var query string
switch store.cfg.Driver {
case DriverMySQL:
query = `SELECT TABLE_NAME, CONSTRAINT_NAME FROM information_schema.TABLE_CONSTRAINTS
WHERE CONSTRAINT_SCHEMA = DATABASE() AND CONSTRAINT_TYPE = 'FOREIGN KEY'
AND TABLE_NAME IN (?, ?)`
case DriverPostgreSQL:
seriesTable, rollupsTable = strings.ToLower(seriesTable), strings.ToLower(rollupsTable)
query = `SELECT table_name, constraint_name FROM information_schema.table_constraints
WHERE table_schema = current_schema() AND constraint_type = 'FOREIGN KEY'
AND table_name IN ($1, $2)`
default:
t.Fatalf("unexpected integration driver %q", store.cfg.Driver)
}
rows, err := store.db.QueryContext(ctx, query, seriesTable, rollupsTable)
if err != nil {
t.Fatalf("list %s foreign keys: %v", store.cfg.Driver, err)
}
type constraint struct{ table, name string }
constraints := make([]constraint, 0, 4)
for rows.Next() {
var item constraint
if err := rows.Scan(&item.table, &item.name); err != nil {
_ = rows.Close()
t.Fatalf("scan %s foreign key: %v", store.cfg.Driver, err)
}
constraints = append(constraints, item)
}
if err := rows.Err(); err != nil {
_ = rows.Close()
t.Fatalf("iterate %s foreign keys: %v", store.cfg.Driver, err)
}
if err := rows.Close(); err != nil {
t.Fatalf("close %s foreign keys: %v", store.cfg.Driver, err)
}
if len(constraints) != 4 {
t.Fatalf("%s foreign keys before removal = %d, want 4", store.cfg.Driver, len(constraints))
}
clause := "DROP CONSTRAINT"
if store.cfg.Driver == DriverMySQL {
clause = "DROP FOREIGN KEY"
}
for _, item := range constraints {
statement := fmt.Sprintf("ALTER TABLE %s %s %s",
quoteMaintenanceIdentifier(store.cfg.Driver, item.table), clause,
quoteMaintenanceIdentifier(store.cfg.Driver, item.name))
if _, err := store.db.ExecContext(ctx, statement); err != nil {
t.Fatalf("drop %s foreign key %s: %v", store.cfg.Driver, item.name, err)
}
}
}
func assertIntegrationForeignKeys(t *testing.T, store *Store) {
t.Helper()
var query string
switch store.cfg.Driver {
case DriverMySQL:
query = `SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS
WHERE CONSTRAINT_SCHEMA = DATABASE() AND CONSTRAINT_TYPE = 'FOREIGN KEY'
AND TABLE_NAME IN (?, ?)`
case DriverPostgreSQL:
query = `SELECT COUNT(*) FROM information_schema.table_constraints
WHERE table_schema = current_schema() AND constraint_type = 'FOREIGN KEY'
AND table_name IN ($1, $2)`
default:
t.Fatalf("unexpected integration driver %q", store.cfg.Driver)
}
var count int
seriesTable, rollupsTable := store.tables.series, store.tables.rollups
if store.cfg.Driver == DriverPostgreSQL {
seriesTable, rollupsTable = strings.ToLower(seriesTable), strings.ToLower(rollupsTable)
}
if err := store.db.QueryRow(query, seriesTable, rollupsTable).Scan(&count); err != nil {
t.Fatalf("query foreign key metadata: %v", err)
}
if count != 4 {
t.Fatalf("foreign key count = %d, want 4", count)
}
}
// dropIntegrationTables drops integration-test tables.
//
// dropIntegrationTables 删除集成测试创建的表。
func dropIntegrationTables(t *testing.T, store *Store, prefix string) {
t.Helper()
if strings.TrimSpace(prefix) == "" {
t.Fatal("refusing to drop tables with empty prefix")
}
for _, name := range []string{
prefix + "rebuild_points",
prefix + "rebuild_rollups",
prefix + "rebuild_series",
prefix + "rebuild_label_sets",
prefix + "rebuild_resolutions",
prefix + "rebuild_definitions",
prefix + "rebuild_store_state",
prefix + "compaction_watermarks_legacy",
prefix + "points_legacy",
prefix + "rollups_legacy",
prefix + "series_legacy",
prefix + "label_sets_legacy",
prefix + "resolutions_legacy",
prefix + "definitions_legacy",
prefix + "compaction_watermarks",
prefix + "points",
prefix + "rollups",
prefix + "series",
prefix + "label_sets",
prefix + "resolutions",
prefix + "definitions",
prefix + "store_state",
} {
if _, err := store.db.Exec("DROP TABLE IF EXISTS " + name); err != nil {
t.Fatalf("drop integration table %s: %v", name, err)
}
}
}
-15
View File
@@ -8,21 +8,6 @@ import (
"time"
)
func TestNetworkTestMethodConstants(t *testing.T) {
tests := map[string]string{
MethodNetworkTestNextTrace: "networkTest.nextTrace",
MethodNetworkTestIperf3: "networkTest.iperf3",
MethodNetworkTestMeshTrace: "networkTest.meshTrace",
MethodNetworkTestGetMeshTraceJob: "networkTest.getMeshTraceJob",
}
for got, want := range tests {
if got != want {
t.Fatalf("method constant = %q, want %q", got, want)
}
}
}
func TestNextTraceParamsJSON(t *testing.T) {
params := NextTraceParams{
TaskID: "trace-1",
+2 -61
View File
@@ -1,72 +1,13 @@
package geoip_test
import (
"net"
"testing"
"github.com/komari-monitor/komari/utils/geoip"
)
// 测试GeoIP数据库的初始化和更新功能
func TestMmdb(t *testing.T) {
geoip.CurrentProvider, _ = geoip.NewMaxMindGeoIPService()
testIpAddr(t)
}
func TestIpApi(t *testing.T) {
geoip.CurrentProvider, _ = geoip.NewIPAPIService()
testIpAddr(t)
}
func TestGeojs(t *testing.T) {
geoip.CurrentProvider, _ = geoip.NewGeoJSService()
testIpAddr(t)
}
func TestIpInfo(t *testing.T) {
geoip.CurrentProvider, _ = geoip.NewIPInfoService()
testIpAddr(t)
}
func testIpAddr(t *testing.T) {
// IPv4
ipaddr := "8.8.8.8"
ip := net.ParseIP(ipaddr)
record, err := geoip.GetGeoInfo(ip)
if err != nil {
t.Errorf("Failed to get GeoIP info for IP %s: %v", ipaddr, err)
}
if record != nil {
if record.ISOCode == "" && record.Name == "" {
t.Errorf("Country information is missing for IP %s", ipaddr)
}
} else {
t.Errorf("GeoIP record is nil for IP %s", ipaddr)
}
t.Logf("IPv4:[%s]%s - %s", ipaddr, record.ISOCode, record.Name)
// IPv6
ipaddr = "2001:4860:4860::8888"
ip = net.ParseIP(ipaddr)
record, err = geoip.GetGeoInfo(ip)
if err != nil {
t.Errorf("Failed to get GeoIP info for IPv6 %s: %v", ipaddr, err)
}
if record != nil {
if record.ISOCode == "" && record.Name == "" {
t.Errorf("Country information is missing for IPv6 %s", ipaddr)
}
} else {
t.Errorf("GeoIP record is nil for IPv6 %s", ipaddr)
}
t.Logf("IPv6:[%s]%s - %s", ipaddr, record.ISOCode, record.Name)
}
func TestUnicodeEmoji(t *testing.T) {
ISOCode := "CN"
emoji := geoip.GetRegionUnicodeEmoji(ISOCode)
if emoji != "🇨🇳" {
t.Errorf("Expected emoji for %s, got %s", ISOCode, emoji)
if emoji := geoip.GetRegionUnicodeEmoji("CN"); emoji != "🇨🇳" {
t.Errorf("Expected emoji for CN, got %s", emoji)
}
t.Logf("Emoji for %s: %s", ISOCode, emoji)
}
+23 -58
View File
@@ -1,4 +1,4 @@
package geoip // 与 geoip.go 保持相同的包名,表示它们是同一个包的组成部分
package geoip
import (
"fmt"
@@ -7,21 +7,16 @@ import (
"net"
"net/http"
"os"
"path/filepath" // 新增导入,用于处理文件路径
"path/filepath"
"sync"
"github.com/komari-monitor/komari/database/auditlog"
"github.com/oschwald/maxminddb-golang"
)
// GeoIpUrl 是 MaxMind 数据库的下载地址。
var GeoIpUrl = "https://raw.githubusercontent.com/Loyalsoldier/geoip/release/GeoLite2-Country.mmdb"
// GeoIpFilePath 是本地存储 MaxMind 数据库的路径。
var GeoIpFilePath = "./data/GeoLite2-Country.mmdb"
// GeoIpRecord 结构体定义了 MaxMind 数据库查询结果的原始结构。
// 它是 MaxMind 库特有的,用于从 .mmdb 文件中解析数据。
type GeoIpRecord struct {
Country struct {
ISOCode string `maxminddb:"iso_code"`
@@ -29,37 +24,27 @@ type GeoIpRecord struct {
} `maxminddb:"country"`
}
// MaxMindGeoIPService 是 GeoIPService 接口的一个具体实现,
// 它使用 MaxMind 数据库作为后端。
type MaxMindGeoIPService struct {
// maxMindDBReader 内部持有的 MaxMind 数据库读取器实例。
maxMindDBReader *maxminddb.Reader
// dbFilePath 是 MaxMind 数据库文件的路径。
dbFilePath string
// mu 用于保护对 maxMindDBReader 的并发访问,确保线程安全。
mu sync.RWMutex
dbFilePath string
mu sync.RWMutex
}
// Name 返回服务的名称。
func (s *MaxMindGeoIPService) Name() string {
return "MaxMind"
}
// NewMaxMindGeoIPService 创建并返回一个 MaxMindGeoIPService 实例。
// 它负责初始化服务,包括尝试加载或下载数据库。
func NewMaxMindGeoIPService() (*MaxMindGeoIPService, error) {
dbFilePath := GeoIpFilePath
service := &MaxMindGeoIPService{
dbFilePath: dbFilePath,
}
// 确保数据目录存在
if err := os.MkdirAll(filepath.Dir(dbFilePath), os.ModePerm); err != nil {
auditlog.Log("", "", "Failed to create data directory for MaxMind database: "+err.Error(), "error")
return nil, fmt.Errorf("failed to create data directory for MaxMind database: %w", err)
}
// 检查数据库文件是否存在,如果不存在则尝试下载
if _, err := os.Stat(dbFilePath); os.IsNotExist(err) {
if err := service.UpdateDatabase(); err != nil {
auditlog.Log("", "", "Failed to download initial MaxMind database: "+err.Error(), "error")
@@ -67,7 +52,6 @@ func NewMaxMindGeoIPService() (*MaxMindGeoIPService, error) {
}
}
// 初始化或重新加载 MaxMind 数据库。
if err := service.initialize(); err != nil {
auditlog.Log("", "", "Failed to initialize MaxMind database: "+err.Error(), "error")
return nil, fmt.Errorf("failed to initialize MaxMind database: %w", err)
@@ -75,20 +59,15 @@ func NewMaxMindGeoIPService() (*MaxMindGeoIPService, error) {
return service, nil
}
// initialize 初始化或重新加载 MaxMind 数据库。
// 这是一个内部方法,供 NewMaxMindGeoIPService 和 UpdateDatabase 调用。
// 它会关闭现有连接(如果存在)并重新打开数据库文件。
func (s *MaxMindGeoIPService) initialize() error {
s.mu.Lock()
defer s.mu.Unlock()
// 如果已存在数据库读取器,则先关闭它。
if s.maxMindDBReader != nil {
s.maxMindDBReader.Close()
s.maxMindDBReader = nil
}
// 尝试打开新的数据库文件。
reader, err := maxminddb.Open(s.dbFilePath)
if err != nil {
return fmt.Errorf("error opening MaxMind database at %s: %w", s.dbFilePath, err)
@@ -97,8 +76,6 @@ func (s *MaxMindGeoIPService) initialize() error {
return nil
}
// GetGeoInfo 根据 IP 地址获取 MaxMind 的地理位置信息。
// 它查询 MaxMind 数据库并将其特有的 GeoIpRecord 转换为通用的 GeoInfo 结构体。
func (s *MaxMindGeoIPService) GetGeoInfo(ip net.IP) (*GeoInfo, error) {
s.mu.RLock()
defer s.mu.RUnlock()
@@ -106,72 +83,60 @@ func (s *MaxMindGeoIPService) GetGeoInfo(ip net.IP) (*GeoInfo, error) {
if s.maxMindDBReader == nil {
return nil, fmt.Errorf("MaxMind database is not initialized or failed to open")
}
if ip == nil {
return nil, fmt.Errorf("IP address cannot be nil")
}
var record GeoIpRecord // 使用原始的 GeoIpRecord 结构体来接收查询结果
err := s.maxMindDBReader.Lookup(ip, &record)
if err != nil {
// 返回错误,但避免直接返回 maxminddb 库的内部错误,提供更友好的信息
var record GeoIpRecord
if err := s.maxMindDBReader.Lookup(ip, &record); err != nil {
return nil, fmt.Errorf("error looking up IP %s in MaxMind database: %w", ip.String(), err)
}
// 将 MaxMind 的特定结构体转换为通用的 GeoInfo 结构体
geoInfo := &GeoInfo{
ISOCode: record.Country.ISOCode,
// 尝试获取英文国家名称,如果不存在则使用 ISO 代码作为备用
Name: record.Country.Names["en"],
Name: record.Country.Names["en"],
}
if geoInfo.Name == "" && geoInfo.ISOCode != "" {
geoInfo.Name = geoInfo.ISOCode // 如果没有英文名称,回退到 ISO 代码
geoInfo.Name = geoInfo.ISOCode
}
return geoInfo, nil
}
// UpdateDatabase 实现了 GeoIPService 接口的 UpdateDatabase 方法。
// 它会下载最新的 GeoLite2-Country.mmdb 文件并重新加载数据库。
func (s *MaxMindGeoIPService) UpdateDatabase() error {
s.mu.Lock() // 获取写锁,确保更新过程的互斥性
if err := s.download(); err != nil {
return err
}
return s.initialize()
}
resp, err := http.Get(GeoIpUrl) // GeoIpUrl 是预定义的 MaxMind 数据库下载地址
// download 下载最新数据库文件,持有写锁避免与查询并发
func (s *MaxMindGeoIPService) download() error {
s.mu.Lock()
defer s.mu.Unlock()
resp, err := http.Get(GeoIpUrl)
if err != nil {
return fmt.Errorf("failed to initiate MaxMind database download: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("failed to download MaxMind database: HTTP status %s", resp.Status)
}
// 确保数据目录存在(NewMaxMindGeoIPService 已处理,但这里再次确保以防直接调用)
if err := os.MkdirAll(filepath.Dir(s.dbFilePath), os.ModePerm); err != nil {
return fmt.Errorf("failed to create data directory for MaxMind database update: %w", err)
}
out, err := os.Create(s.dbFilePath) // 创建或覆盖本地数据库文件
out, err := os.Create(s.dbFilePath)
if err != nil {
return fmt.Errorf("failed to create MaxMind database file at %s: %w", s.dbFilePath, err)
}
defer out.Close()
_, err = io.Copy(out, resp.Body) // 将下载内容写入文件
if err != nil {
if _, err := io.Copy(out, resp.Body); err != nil {
return fmt.Errorf("failed to write MaxMind database file: %w", err)
}
s.mu.Unlock() // initialize 方法需要在解锁后调用,以避免死锁
// 重新加载数据库以使用新下载的文件
return s.initialize()
return nil
}
// Close 实现了 GeoIPService 接口的 Close 方法。
// 它关闭 MaxMind 数据库读取器,释放文件句柄和其他资源。
func (s *MaxMindGeoIPService) Close() error {
s.mu.Lock()
defer s.mu.Unlock()
if s.maxMindDBReader != nil {
err := s.maxMindDBReader.Close()
s.maxMindDBReader = nil // 清空读取器实例
s.maxMindDBReader = nil
if err != nil {
return fmt.Errorf("error closing MaxMind database: %w", err)
}
-4
View File
@@ -13,10 +13,6 @@ import (
"github.com/komari-monitor/komari/utils/renewal"
)
func CheckExpireScheduledWork() {
CheckExpire()
}
func CheckExpire() {
cfg, err := config.GetMany(map[string]any{
config.ExpireNotificationEnabledKey: false,
+7 -32
View File
@@ -39,13 +39,7 @@ func CheckTraffic() {
// 起始阈值:例如 80%,非5的倍数则从上取整到最近的5的倍数,例如 83->85
startThreshold := cfg
if startThreshold < 0 {
startThreshold = 0
}
baseStep := int(math.Ceil(startThreshold/5.0) * 5.0)
if baseStep > 100 {
baseStep = 100
}
baseStep := min(int(math.Ceil(startThreshold/5.0)*5.0), 100)
allClients, err := clients.GetAllClientBasicInfo()
if err != nil {
@@ -58,7 +52,7 @@ func CheckTraffic() {
}
r, ok := reports[c.UUID]
if !ok || r == nil {
if !ok {
continue
}
@@ -74,13 +68,7 @@ func CheckTraffic() {
}
// 当前所在阈值步进(5%的倍数)
curStep := int(math.Floor(pct/5.0) * 5.0)
if curStep < baseStep {
curStep = baseStep
}
// if curStep > 100 {
// curStep = 100
// }
curStep := max(int(math.Floor(pct/5.0)*5.0), baseStep)
key := "traffic:" + c.UUID
last, _ := trafficCache.Get(key)
@@ -116,17 +104,9 @@ func computeUsedByType(t string, up, down int64) int64 {
case "sum":
return up + down
case "min":
if up < down {
return up
}
return down
case "max":
fallthrough
default:
if up > down {
return up
}
return down
return min(up, down)
default: // "max"
return max(up, down)
}
}
@@ -140,10 +120,5 @@ func humanBytes(b int64) string {
div *= unit
exp++
}
// KMGTPE
prefixes := []string{"K", "M", "G", "T", "P", "E"}
if exp >= len(prefixes) {
exp = len(prefixes) - 1
}
return fmt.Sprintf("%.2f %sB", float64(b)/float64(div), prefixes[exp])
return fmt.Sprintf("%.2f %cB", float64(b)/float64(div), "KMGTPE"[exp])
}
+1 -6
View File
@@ -5,9 +5,6 @@ import (
)
func GenerateRandomString(length int) string {
if length <= 0 {
return ""
}
const charset = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
const n = byte(len(charset)) // 62
const threshold = byte(256 - (256 % len(charset))) // 248
@@ -15,9 +12,7 @@ func GenerateRandomString(length int) string {
out := make([]byte, 0, length)
buf := make([]byte, length)
for len(out) < length {
if _, err := rand.Read(buf); err != nil {
return ""
}
rand.Read(buf)
for _, b := range buf {
if b < threshold {
out = append(out, charset[int(b%n)])
+13 -60
View File
@@ -14,17 +14,9 @@ import (
)
func CheckAndAutoRenewal(client models.Client) {
// 自动续费检查
//type renewedClient struct {
// Name string
// NewExpireTime time.Time
//}
//var renewedClients []renewedClient
if !client.AutoRenewal {
return
}
// 不在线则不续费
if _, ok := agent_runtime.GetConnectedClients()[client.UUID]; !ok {
return
}
@@ -35,78 +27,54 @@ func CheckAndAutoRenewal(client models.Client) {
clientExpireTime := client.ExpiredAt.UTC()
checkTime := time.Now().UTC()
// 如果到期时间小于0002年,跳过
if clientExpireTime.Year() < 2 {
return
}
// 检查是否已过期或当天过期
if clientExpireTime.Before(checkTime) || timeutil.SameSystemDate(clientExpireTime, checkTime) {
// 计算过期时间距离创建时间的总天数,判断是否为长期账单
now := checkTime
localNow := now.In(time.Local)
hundredYearsFromNow := localNow.AddDate(100, 0, 0).UTC()
// 如果过期时间超过当前时间100年,视为长期/一次性账单,不续费
if clientExpireTime.After(hundredYearsFromNow) {
localNow := checkTime.In(time.Local)
if clientExpireTime.After(localNow.AddDate(100, 0, 0).UTC()) {
return
}
// 如果有账单周期且不为0,进行自动续费
if client.BillingCycle > 0 {
// 根据账单周期计算新的过期时间
var newExpireTime time.Time
billingCycle := client.BillingCycle
// 如果服务器的过期时间太早了,那么直接设置为从当前时间算的下一个到期时间
baseTime := clientExpireTime.In(time.Local)
if clientExpireTime.Before(localNow.AddDate(0, 0, -30).UTC()) { // 过期时间超过30天前
if clientExpireTime.Before(localNow.AddDate(0, 0, -30).UTC()) {
baseTime = localNow
}
if billingCycle >= 27 && billingCycle <= 32 {
// 月度计费 - 加1个月
switch {
case billingCycle >= 27 && billingCycle <= 32:
newExpireTime = baseTime.AddDate(0, 1, 0)
} else if billingCycle >= 87 && billingCycle <= 95 {
// 季度计费 - 加3个月
case billingCycle >= 87 && billingCycle <= 95:
newExpireTime = baseTime.AddDate(0, 3, 0)
} else if billingCycle >= 175 && billingCycle <= 185 {
// 半年计费 - 加6个月
case billingCycle >= 175 && billingCycle <= 185:
newExpireTime = baseTime.AddDate(0, 6, 0)
} else if billingCycle >= 360 && billingCycle <= 370 {
// 年度计费 - 加1年
case billingCycle >= 360 && billingCycle <= 370:
newExpireTime = baseTime.AddDate(1, 0, 0)
} else if billingCycle >= 720 && billingCycle <= 750 {
// 两年计费 - 加2年
case billingCycle >= 720 && billingCycle <= 750:
newExpireTime = baseTime.AddDate(2, 0, 0)
} else if billingCycle >= 1080 && billingCycle <= 1150 {
// 三年计费 - 加3年
case billingCycle >= 1080 && billingCycle <= 1150:
newExpireTime = baseTime.AddDate(3, 0, 0)
} else if billingCycle >= 1800 && billingCycle <= 1850 {
// 五年计费 - 加5年
case billingCycle >= 1800 && billingCycle <= 1850:
newExpireTime = baseTime.AddDate(5, 0, 0)
} else {
// 其他情况,直接加上账单周期天数
default:
newExpireTime = baseTime.AddDate(0, 0, billingCycle)
}
// 更新客户端过期时间
updates := map[string]interface{}{
"uuid": client.UUID,
"expired_at": newExpireTime.UTC(),
}
err := clients.SaveClient(updates)
if err != nil {
if err := clients.SaveClient(updates); err != nil {
auditlog.EventLog("renewal", fmt.Sprintf("Failed to renew client %s (%s): %v", client.Name, client.UUID, err))
return
}
//renewedClients = append(renewedClients, renewedClient{
// Name: client.Name,
// NewExpireTime: newExpireTime,
//})
auditlog.EventLog("renewal", fmt.Sprintf("Auto-renewed client: %s until %s",
client.Name, timeutil.FormatSystemDate(newExpireTime)))
@@ -119,19 +87,4 @@ func CheckAndAutoRenewal(client models.Client) {
})
}
}
// 发送续费通知
// if len(renewedClients) > 0 {
// message := ""
// for _, clientInfo := range renewedClients {
// message += fmt.Sprintf("• %s until %s\n", clientInfo.Name, clientInfo.NewExpireTime.Format("2006-01-02"))
// }
// messageSender.SendEvent(models.EventMessage{
// Event: messageevent.Renew,
// Clients: []models.Client{client},
// Time: time.Now(),
// Emoji: "🔄",
// Message: message,
// })
// }
}
+2 -35
View File
@@ -31,17 +31,10 @@ func DataMasking(str string, private []string) string {
uniq[w] = struct{}{}
words = append(words, w)
}
if len(words) == 0 {
return str
}
// 逐词进行滑动窗口匹配 + 模糊匹配(Levenshtein 相似度)
for _, w := range words {
wRunes := []rune(w)
wl := len(wRunes)
if wl == 0 || wl > n {
continue
}
// 滑动窗口大小采用敏感词长度
for i := 0; i <= n-wl; i++ {
@@ -102,25 +95,12 @@ func similarity(a, b string) float64 {
ar := []rune(a)
br := []rune(b)
dist := levenshtein(ar, br)
maxLen := len(ar)
if len(br) > maxLen {
maxLen = len(br)
}
if maxLen == 0 {
return 1
}
return 1 - float64(dist)/float64(maxLen)
return 1 - float64(dist)/float64(max(len(ar), len(br)))
}
// levenshtein 计算两个 rune slice 的编辑距离
func levenshtein(a, b []rune) int {
la, lb := len(a), len(b)
if la == 0 {
return lb
}
if lb == 0 {
return la
}
// 使用滚动数组降低空间复杂度
prev := make([]int, lb+1)
curr := make([]int, lb+1)
@@ -134,22 +114,9 @@ func levenshtein(a, b []rune) int {
if a[i-1] != b[j-1] {
cost = 1
}
del := prev[j] + 1
ins := curr[j-1] + 1
sub := prev[j-1] + cost
curr[j] = minInt(del, ins, sub)
curr[j] = min(prev[j]+1, curr[j-1]+1, prev[j-1]+cost)
}
prev, curr = curr, prev
}
return prev[lb]
}
func minInt(vals ...int) int {
m := vals[0]
for _, v := range vals[1:] {
if v < m {
m = v
}
}
return m
}
+2 -8
View File
@@ -1,6 +1,7 @@
package agent
import (
"maps"
"sort"
"sync"
"time"
@@ -28,11 +29,7 @@ const recentReportRetention = time.Minute
func GetConnectedClients() map[string]*connection.SafeConn {
mu.RLock()
defer mu.RUnlock()
clientsCopy := make(map[string]*connection.SafeConn)
for k, v := range connectedClients {
clientsCopy[k] = v
}
return clientsCopy
return maps.Clone(connectedClients)
}
func SetConnectedClients(uuid string, conn *connection.SafeConn) {
@@ -127,9 +124,6 @@ func GetLatestReport() map[string]*v2.Report {
defer mu.RUnlock()
reportCopy := make(map[string]*v2.Report)
for k, v := range latestReport {
if v == nil {
continue
}
item := *v
reportCopy[k] = &item
}
+12 -50
View File
@@ -2,7 +2,6 @@ package api
import (
"bytes"
"database/sql"
"encoding/json"
"io"
"net/http"
@@ -11,10 +10,8 @@ import (
"github.com/gin-gonic/gin"
"github.com/komari-monitor/komari/database/accounts"
"github.com/komari-monitor/komari/database/clients"
"github.com/komari-monitor/komari/internal/config"
"github.com/komari-monitor/komari/pkg/rpc"
"gorm.io/gorm"
)
const (
@@ -71,12 +68,8 @@ func RequireRole(allowedRoles ...string) gin.HandlerFunc {
// GetRole 获取当前请求的角色
func GetRole(c *gin.Context) string {
role, exists := c.Get("role")
if !exists {
return RoleGuest
}
if s, ok := role.(string); ok {
return s
if role := c.GetString("role"); role != "" {
return role
}
return RoleGuest
}
@@ -151,23 +144,16 @@ func hasTempAccess(c *gin.Context) bool {
if err != nil {
return false
}
expireAt, err := config.GetAs[int64]("tempory_share_token_expire_at", 0)
if err != nil {
return false
}
allowKey, err := config.GetAs[string]("tempory_share_token", "")
if err != nil {
if err != nil || allowKey == "" || tempKey != allowKey {
return false
}
if allowKey == "" || tempKey != allowKey {
return false
}
return expireAt >= time.Now().Unix()
expireAt, err := config.GetAs[int64]("tempory_share_token_expire_at", 0)
return err == nil && expireAt >= time.Now().Unix()
}
func extractClientToken(c *gin.Context) string {
token := c.Query("token")
if token != "" {
if token := c.Query("token"); token != "" {
return token
}
// rpc2 约定:agent 经 ?Authorization=<token> 传入 client token。
@@ -182,43 +168,19 @@ func extractClientToken(c *gin.Context) string {
}
c.Request.Body = io.NopCloser(bytes.NewReader(bodyBytes))
var bodyMap map[string]interface{}
if len(bodyBytes) > 0 {
if err := json.Unmarshal(bodyBytes, &bodyMap); err == nil {
if tokenVal, exists := bodyMap["token"]; exists {
if str, ok := tokenVal.(string); ok && str != "" {
return str
}
}
}
var body struct {
Token string `json:"token"`
}
if json.Unmarshal(bodyBytes, &body) == nil {
return body.Token
}
}
return ""
}
func checkTokenAndGetUUID(token string) (string, error) {
uuid, err := clients.GetClientUUIDByToken(token)
if err == sql.ErrNoRows {
return "", nil
}
if err == gorm.ErrRecordNotFound {
return "", nil
}
if err != nil {
return "", err
}
return uuid, nil
}
func isApiKeyValid(apiKey string) bool {
apiKeyConfig, err := config.GetAs[string](config.ApiKeyKey, "")
if err != nil {
return false
}
if apiKeyConfig == "" || len(apiKeyConfig) < 12 {
if err != nil || len(apiKeyConfig) < 12 {
return false
}
return apiKey == "Bearer "+apiKeyConfig
+2 -3
View File
@@ -17,14 +17,13 @@ func RequireSensitive2FA() gin.HandlerFunc {
c.Abort()
return
}
// 标记本请求已通过敏感操作校验,避免下游(RPC 边界)重复校验。
// 标记已通过校验,避免 RPC 边界重复校验
c.Set("sensitive_2fa_verified", true)
c.Next()
}
}
// VerifySensitive2FACore 传输无关的 2FA 校验核心。
// 输入原始值:userUUID、2FA code、是否为 API Key。
// API Key 豁免;未启用 2FA 的用户放行;其余需要有效 code。
func VerifySensitive2FACore(userUUID, code string, isAPIKey bool) error {
if isAPIKey {
@@ -53,7 +52,7 @@ func VerifySensitive2FACore(userUUID, code string, isAPIKey bool) error {
return nil
}
// VerifySensitive2FA gin 适配层:从 gin.Context 提取参数后委托核心校验。
// VerifySensitive2FA gin 适配层,从 Context 提取参数后委托核心校验。
func VerifySensitive2FA(c *gin.Context) error {
_, isAPIKey := c.Get("api_key")
uuidRaw, _ := c.Get("uuid")
-14
View File
@@ -50,17 +50,3 @@ func TestPluginMarketI18nTextAndSourceOnlyEntry(t *testing.T) {
t.Fatalf("validatePluginMarketPlugin() error = %v", err)
}
}
func TestDefaultPluginMarketSourcePointsToOfficialRepo(t *testing.T) {
sources := defaultPluginMarketSources()
if len(sources) != 1 {
t.Fatalf("defaultPluginMarketSources() = %#v, want exactly one source", sources)
}
source := sources[0]
if source.ID != "official" || !source.Enabled {
t.Fatalf("defaultPluginMarketSources() = %#v, want enabled official source", source)
}
if source.URL != "https://raw.githubusercontent.com/komari-monitor/plugin-market/main/v1.json" {
t.Fatalf("default plugin market URL = %q", source.URL)
}
}
+1 -8
View File
@@ -10,19 +10,12 @@ import (
func RegisterClient(c *gin.Context) {
auth := c.GetHeader("Authorization")
if auth == "" {
api.RespondError(c, 403, "Invalid AutoDiscovery Key")
return
}
AutoDiscoveryKey, err := config.GetAs[string](config.AutoDiscoveryKeyKey, "")
if err != nil {
api.RespondError(c, 500, "Failed to get AutoDiscovery Key: "+err.Error())
return
}
if AutoDiscoveryKey == "" ||
len(AutoDiscoveryKey) < 12 ||
"Bearer "+AutoDiscoveryKey != auth {
if len(AutoDiscoveryKey) < 12 || "Bearer "+AutoDiscoveryKey != auth {
api.RespondError(c, 403, "Invalid AutoDiscovery Key")
return
}
+2 -3
View File
@@ -44,10 +44,9 @@ func refreshPostPresence(uuid string) {
agent_runtime.MarkV2Client(uuid)
go notifier.OnlineNotification(uuid, connID)
defaultGeneration := uint64(0)
entry := &postPresenceEntry{connID: connID, generation: defaultGeneration}
entry := &postPresenceEntry{connID: connID}
entry.timer = time.AfterFunc(postPresenceTTL, func() {
postPresenceExpired(uuid, connID, defaultGeneration)
postPresenceExpired(uuid, connID, 0)
})
postPresenceStates[uuid] = entry
}
+1 -5
View File
@@ -247,13 +247,9 @@ func clientUUIDFromContext(c *gin.Context) (string, bool) {
}
func notifierOnline(uuid string, connID int64) {
go func() {
defer func() { _ = recover() }()
notifier.OnlineNotification(uuid, connID)
}()
go notifier.OnlineNotification(uuid, connID)
}
func notifierOffline(uuid string, connID int64) {
defer func() { _ = recover() }()
notifier.OfflineNotification(uuid, connID)
}
+1 -2
View File
@@ -15,9 +15,8 @@ func getClientIPType(ip net.IP) int {
}
if ip.To4() == nil {
return 1
} else {
return 0
}
return 0
}
func saveClientBasicInfo(info map[string]interface{}, uuid string, fallbackIP string) error {
+5 -7
View File
@@ -8,6 +8,7 @@ package api
import (
"github.com/gin-gonic/gin"
"github.com/komari-monitor/komari/database/accounts"
"github.com/komari-monitor/komari/database/clients"
"github.com/komari-monitor/komari/pkg/rpc"
)
@@ -31,7 +32,7 @@ func IdentifyPrincipal(c *gin.Context) *rpc.Principal {
// 3. Client Token(agent 客户端)
if token := extractClientToken(c); token != "" {
if uuid, err := checkTokenAndGetUUID(token); err == nil && uuid != "" {
if uuid, err := clients.GetClientUUIDByToken(token); err == nil {
return rpc.NewAgentPrincipal(uuid)
}
}
@@ -47,10 +48,7 @@ func SetPrincipal(c *gin.Context, p *rpc.Principal) {
// GetPrincipal 读取已识别的主体;若中间件未运行则返回 nil。
func GetPrincipal(c *gin.Context) *rpc.Principal {
if v, ok := c.Get(principalContextKey); ok {
if p, ok := v.(*rpc.Principal); ok {
return p
}
}
return nil
p, _ := c.Get(principalContextKey)
principal, _ := p.(*rpc.Principal)
return principal
}
+1 -9
View File
@@ -1,8 +1,6 @@
package public
import (
"encoding/json"
"io"
"net/http"
"github.com/komari-monitor/komari/database/accounts"
@@ -41,14 +39,8 @@ func Login(c *gin.Context) {
return
}
bodyBytes, err := io.ReadAll(c.Request.Body)
if err != nil {
api.RespondError(c, http.StatusBadRequest, "Invalid request body: "+err.Error())
return
}
var data LoginRequest
err = json.Unmarshal(bodyBytes, &data)
if err != nil {
if err := c.ShouldBindJSON(&data); err != nil {
api.RespondError(c, http.StatusBadRequest, "Invalid request body: "+err.Error())
return
}
+8 -20
View File
@@ -37,27 +37,16 @@ func OAuthCallback(c *gin.Context) {
// 获取当前OAuth提供商名称
providerName := oauth.CurrentProvider().GetName()
providersSkipStateCheck := []string{"qq"}
if slices.Contains(providersSkipStateCheck, providerName) {
// 对于QQ登录,由于是通过QQ聚合登录平台中转,state可能会不匹配
// 但我们仍然需要验证state的存在性(不能是空的)
if state == "" {
c.JSON(400, gin.H{"status": "error", "error": "Invalid state"})
return
}
} else {
// 对于其他提供商,严格验证state匹配
if state == "" || state != c.Query("state") {
c.JSON(400, gin.H{"status": "error", "error": "Invalid state"})
return
}
// QQ 经聚合平台中转,state 可能不匹配,只校验非空;其他提供商严格校验匹配
skipStateMatch := slices.Contains([]string{"qq"}, providerName)
if state == "" || (!skipStateMatch && state != c.Query("state")) {
c.JSON(400, gin.H{"status": "error", "error": "Invalid state"})
return
}
queries := make(map[string]string)
for key, values := range c.Request.URL.Query() {
if len(values) > 0 {
queries[key] = values[0]
}
queries[key] = values[0]
}
oidcUser, err := oauth.CurrentProvider().OnCallback(c, state, queries, utils.GetCallbackURL(c))
if err != nil {
@@ -66,7 +55,7 @@ func OAuthCallback(c *gin.Context) {
}
// ID作为SSO ID
sso_id := fmt.Sprintf("%s_%s", oauth.CurrentProvider().GetName(), oidcUser.UserId)
sso_id := fmt.Sprintf("%s_%s", providerName, oidcUser.UserId)
// 如果cookie中有binding_external_account,说明是绑定外部账号
// 否则是登录
@@ -80,8 +69,7 @@ func OAuthCallback(c *gin.Context) {
c.JSON(500, gin.H{"status": "error", "message": "Binding failed"})
return
}
err = accounts.BindingExternalAccount(user.UUID, sso_id)
if err != nil {
if err := accounts.BindingExternalAccount(user.UUID, sso_id); err != nil {
c.JSON(500, gin.H{"status": "error", "message": "Binding failed"})
return
}
-1
View File
@@ -111,5 +111,4 @@ func RequestTerminal(c *gin.Context) {
return
}
conn.WriteMessage(1, []byte("等待被控端连接 waiting for agent...\n"))
//auditlog.Log(c.ClientIP(), userID, "request, terminal id:"+id+",client:"+session.UUID, "terminal")
}
+6 -18
View File
@@ -26,18 +26,10 @@ func GetClients(c *gin.Context) {
}
defer conn.Close()
// 初始化用户信息
var (
isLogin = false
hiddenMap = map[string]bool{}
session, _ = c.Cookie("session_token")
)
// 登录状态检查
session, _ := c.Cookie("session_token")
_, err = accounts.GetUserBySession(session)
if err == nil {
isLogin = true
}
isLogin := err == nil
hiddenMap := map[string]bool{}
// 仅在未登录时需要 Hidden 信息做过滤
if !isLogin {
@@ -51,13 +43,10 @@ func GetClients(c *gin.Context) {
// 请求
for {
var resp struct {
resp := struct {
Online []string `json:"online"` // 已建立连接的客户端uuid列表
Data map[string]v2.Report `json:"data"` // 最后上报的数据
}
resp.Online = []string{}
resp.Data = map[string]v2.Report{}
}{Online: []string{}, Data: map[string]v2.Report{}}
_, data, err := conn.ReadMessage()
if err != nil {
@@ -102,8 +91,7 @@ func GetClients(c *gin.Context) {
resp.Data[key] = *report
}
err = conn.WriteJSON(gin.H{"status": "success", "data": resp})
if err != nil {
if err := conn.WriteJSON(gin.H{"status": "success", "data": resp}); err != nil {
return
}
}
+1 -5
View File
@@ -24,7 +24,7 @@ func (g *Generic) GetConfiguration() factory.Configuration {
func (g *Generic) GetAuthorizationURL(redirectURI string) (string, string) {
state := utils.GenerateRandomString(16)
// 构建GitHub OAuth授权URL
// 构建 OAuth 授权URL
authURL := fmt.Sprintf(
"%s?client_id=%s&state=%s&scope=%s&redirect_uri=%s&response_type=code",
g.Addition.AuthURL,
@@ -46,11 +46,7 @@ func (g *Generic) OnCallback(ctx *gin.Context, state string, query map[string]st
if _, ok := g.stateCache.Get(state); !ok {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
if state == "" {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
// 获取code
if code == "" {
return factory.OidcCallback{}, fmt.Errorf("no code provided")
}
-10
View File
@@ -13,10 +13,6 @@ import (
"github.com/patrickmn/go-cache"
)
func init() {
}
func (g *Github) GetName() string {
return "github"
}
@@ -40,19 +36,13 @@ func (g *Github) OnCallback(ctx *gin.Context, state string, query map[string]str
code := query["code"]
// 验证state防止CSRF攻击
// state, _ := c.Cookie("oauth_state")
if g.stateCache == nil {
return factory.OidcCallback{}, fmt.Errorf("state cache not initialized")
}
if _, ok := g.stateCache.Get(state); !ok {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
if state == "" {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
// 获取code
//code := c.Query("code")
if code == "" {
return factory.OidcCallback{}, fmt.Errorf("no code provided")
}
+2 -4
View File
@@ -54,8 +54,7 @@ func LoadProvider(name string, configJson string) error {
if err := json.Unmarshal([]byte(configJson), currentProvider.GetConfiguration()); err != nil {
return fmt.Errorf("failed to unmarshal config for provider %s: %w", name, err)
}
err := currentProvider.Init()
if err != nil {
if err := currentProvider.Init(); err != nil {
return fmt.Errorf("failed to initialize provider %s: %w", name, err)
}
return nil
@@ -96,8 +95,7 @@ func Initialize() error {
LoadProvider("github", "{}")
return nil
}
err = LoadProvider(provider.Name, provider.Addition)
if err != nil {
if err := LoadProvider(provider.Name, provider.Addition); err != nil {
logger.Errorf("oauth", "Failed to load OIDC provider %s: %v", provider.Name, err)
return err
}
+1 -25
View File
@@ -25,7 +25,6 @@ func (q *QQ) GetConfiguration() factory.Configuration {
func (q *QQ) GetAuthorizationURL(redirectURI string) (string, string) {
state := utils.GenerateRandomString(16)
// 构建请求QQ聚合登录平台的URL
requestURL := fmt.Sprintf(
"%s/connect.php?act=login&appid=%s&appkey=%s&type=%s&redirect_uri=%s",
q.Addition.AggregationURL,
@@ -35,21 +34,17 @@ func (q *QQ) GetAuthorizationURL(redirectURI string) (string, string) {
url.QueryEscape(redirectURI),
)
// 向聚合登录平台发送请求
resp, err := http.Get(requestURL)
if err != nil {
// 如果请求失败,返回错误信息
return "", state
}
defer resp.Body.Close()
// 读取响应内容
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", state
}
// 解析响应JSON
var result struct {
Code int `json:"code"`
Msg string `json:"msg"`
@@ -60,7 +55,6 @@ func (q *QQ) GetAuthorizationURL(redirectURI string) (string, string) {
return "", state
}
// 检查响应状态
if result.Code != 0 {
return "", state
}
@@ -69,37 +63,25 @@ func (q *QQ) GetAuthorizationURL(redirectURI string) (string, string) {
return result.URL, state
}
// OnCallback 处理QQ OAuth回调
// 例如:http://localhost:25774/api/oauth_callback?type=qq&code=XXXXXXXXXXXXXXXX
// 然后我们使用code参数向聚合登录平台请求用户信息
// OnCallback 处理 QQ OAuth 回调
func (q *QQ) OnCallback(ctx *gin.Context, state string, query map[string]string, callbackURI string) (factory.OidcCallback, error) {
// 根据文档,回调地址会附带type和code参数
code := query["code"]
loginType := query["type"]
// 如果回调中没有type参数,则使用配置中的LoginType
if loginType == "" {
loginType = q.Addition.LoginType
}
// 验证state防止CSRF攻击
if q.stateCache == nil {
return factory.OidcCallback{}, fmt.Errorf("state cache not initialized")
}
if _, ok := q.stateCache.Get(state); !ok {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
if state == "" {
return factory.OidcCallback{}, fmt.Errorf("invalid state")
}
// 检查是否提供了Authorization Code
if code == "" {
return factory.OidcCallback{}, fmt.Errorf("no authorization code provided")
}
// 通过Authorization Code获取用户信息
// 根据文档,请求URL应为: {AggregationURL}/connect.php?act=callback&appid={appid}&appkey={appkey}&type={登录方式}&code={code}
callbackURL := fmt.Sprintf(
"%s/connect.php?act=callback&appid=%s&appkey=%s&type=%s&code=%s",
q.Addition.AggregationURL,
@@ -115,18 +97,15 @@ func (q *QQ) OnCallback(ctx *gin.Context, state string, query map[string]string,
}
defer resp.Body.Close()
// 检查HTTP响应状态
if resp.StatusCode != http.StatusOK {
return factory.OidcCallback{}, fmt.Errorf("HTTP request failed with status code: %d", resp.StatusCode)
}
// 读取响应
body, err := io.ReadAll(resp.Body)
if err != nil {
return factory.OidcCallback{}, fmt.Errorf("failed to read response: %v", err)
}
// 解析响应
var result struct {
Code int `json:"code"`
Msg string `json:"msg"`
@@ -144,17 +123,14 @@ func (q *QQ) OnCallback(ctx *gin.Context, state string, query map[string]string,
return factory.OidcCallback{}, fmt.Errorf("failed to parse callback response: %v, response body: %s", err, string(body))
}
// 检查返回状态码
if result.Code != 0 {
return factory.OidcCallback{}, fmt.Errorf("QQ login callback failed with code %d: %s", result.Code, result.Msg)
}
// 检查是否返回了用户唯一标识
if result.SocialUid == "" {
return factory.OidcCallback{}, fmt.Errorf("empty social_uid returned, full response: %s", string(body))
}
// 返回用户唯一标识
return factory.OidcCallback{UserId: result.SocialUid}, nil
}