618399f86d
- Replace single Load() with a watching Store that re-reads config files
(and config.local.yaml) and fans out a ReloadEvent with a per-field diff
so consumers can decide whether the change is hot-applicable or requires
a process restart.
- Wrap llm, retrieval, vector store, and object storage clients in
Reloadable* shells so the bootstrap can swap their underlying impls when
config changes without re-instantiating handlers.
- Make jwt.Manager and ops TokenIssuer mutable under a lock so secrets and
TTLs can be rotated live; thread default plan code through a setter on
the ops AdminUserService.
- Wire ConfigStore through bootstrap and every cmd/main.go, scheduler /
worker / tenant-api / ops-api start the watcher; services and handlers
take a config.Provider so they always read current values for things
like generation.stream_enabled, scheduler dispatch, retrieval, etc.
- Switch shared/config decoding off viper to a Kratos-derived runtime
package so env placeholders (\${VAR:default}) resolve consistently and
the same source machinery powers both the loader and the watcher.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
282 lines
7.1 KiB
Go
282 lines
7.1 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/golang-jwt/jwt/v5"
|
|
"github.com/google/uuid"
|
|
"golang.org/x/crypto/bcrypt"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/ops/domain"
|
|
"github.com/geo-platform/tenant-api/internal/ops/repository"
|
|
sharedauth "github.com/geo-platform/tenant-api/internal/shared/auth"
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
)
|
|
|
|
type OperatorClaims struct {
|
|
OperatorID int64 `json:"operator_id"`
|
|
Username string `json:"username"`
|
|
DisplayName string `json:"display_name"`
|
|
Role string `json:"role"`
|
|
jwt.RegisteredClaims
|
|
}
|
|
|
|
type TokenIssuer struct {
|
|
mu sync.RWMutex
|
|
secret []byte
|
|
accessTTL time.Duration
|
|
}
|
|
|
|
func NewTokenIssuer(secret string, accessTTL time.Duration) *TokenIssuer {
|
|
issuer := &TokenIssuer{}
|
|
issuer.Update(secret, accessTTL)
|
|
return issuer
|
|
}
|
|
|
|
func (t *TokenIssuer) Update(secret string, accessTTL time.Duration) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
t.secret = []byte(secret)
|
|
t.accessTTL = accessTTL
|
|
}
|
|
|
|
func (t *TokenIssuer) AccessTTL() time.Duration {
|
|
_, accessTTL := t.snapshot()
|
|
return accessTTL
|
|
}
|
|
|
|
type IssuedToken struct {
|
|
AccessToken string
|
|
ExpiresAt int64
|
|
}
|
|
|
|
func (t *TokenIssuer) Issue(operator *domain.OperatorAccount) (*IssuedToken, error) {
|
|
secret, accessTTL := t.snapshot()
|
|
now := time.Now()
|
|
exp := now.Add(accessTTL)
|
|
claims := OperatorClaims{
|
|
OperatorID: operator.ID,
|
|
Username: operator.Username,
|
|
DisplayName: operator.DisplayName,
|
|
Role: operator.Role,
|
|
RegisteredClaims: jwt.RegisteredClaims{
|
|
ID: uuid.New().String(),
|
|
Subject: "ops-access",
|
|
IssuedAt: jwt.NewNumericDate(now),
|
|
ExpiresAt: jwt.NewNumericDate(exp),
|
|
},
|
|
}
|
|
signed, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(secret)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("sign ops token: %w", err)
|
|
}
|
|
return &IssuedToken{AccessToken: signed, ExpiresAt: exp.Unix()}, nil
|
|
}
|
|
|
|
func (t *TokenIssuer) Parse(raw string) (*OperatorClaims, error) {
|
|
secret, _ := t.snapshot()
|
|
token, err := jwt.ParseWithClaims(raw, &OperatorClaims{}, func(tok *jwt.Token) (any, error) {
|
|
if _, ok := tok.Method.(*jwt.SigningMethodHMAC); !ok {
|
|
return nil, fmt.Errorf("unexpected signing method: %v", tok.Header["alg"])
|
|
}
|
|
return secret, nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
claims, ok := token.Claims.(*OperatorClaims)
|
|
if !ok || !token.Valid {
|
|
return nil, errors.New("invalid ops token")
|
|
}
|
|
if claims.Subject != "ops-access" {
|
|
return nil, errors.New("wrong token subject")
|
|
}
|
|
return claims, nil
|
|
}
|
|
|
|
func (t *TokenIssuer) snapshot() ([]byte, time.Duration) {
|
|
if t == nil {
|
|
return nil, 0
|
|
}
|
|
t.mu.RLock()
|
|
defer t.mu.RUnlock()
|
|
secret := append([]byte(nil), t.secret...)
|
|
return secret, t.accessTTL
|
|
}
|
|
|
|
type AuthService struct {
|
|
accounts *repository.AccountRepository
|
|
audits *AuditService
|
|
issuer *TokenIssuer
|
|
guard *sharedauth.LoginGuard
|
|
}
|
|
|
|
func NewAuthService(
|
|
accounts *repository.AccountRepository,
|
|
audits *AuditService,
|
|
issuer *TokenIssuer,
|
|
guards ...*sharedauth.LoginGuard,
|
|
) *AuthService {
|
|
var guard *sharedauth.LoginGuard
|
|
if len(guards) > 0 {
|
|
guard = guards[0]
|
|
}
|
|
return &AuthService{accounts: accounts, audits: audits, issuer: issuer, guard: guard}
|
|
}
|
|
|
|
type LoginInput struct {
|
|
Username string
|
|
Password string
|
|
IP string
|
|
UserAgent string
|
|
RequestID string
|
|
}
|
|
|
|
type LoginResult struct {
|
|
AccessToken string
|
|
ExpiresAt int64
|
|
Operator OperatorView
|
|
}
|
|
|
|
type OperatorView struct {
|
|
ID int64 `json:"id"`
|
|
Username string `json:"username"`
|
|
DisplayName string `json:"display_name"`
|
|
Email *string `json:"email"`
|
|
Role string `json:"role"`
|
|
Status string `json:"status"`
|
|
}
|
|
|
|
func ToOperatorView(a *domain.OperatorAccount) OperatorView {
|
|
return OperatorView{
|
|
ID: a.ID,
|
|
Username: a.Username,
|
|
DisplayName: a.DisplayName,
|
|
Email: a.Email,
|
|
Role: a.Role,
|
|
Status: a.Status,
|
|
}
|
|
}
|
|
|
|
const (
|
|
ActionLoginSuccess = "auth.login.success"
|
|
ActionLoginFailed = "auth.login.failure"
|
|
)
|
|
|
|
func (s *AuthService) Login(ctx context.Context, in LoginInput) (*LoginResult, error) {
|
|
in.Username = strings.TrimSpace(in.Username)
|
|
if in.Username == "" {
|
|
return nil, response.ErrBadRequest(40000, "invalid_payload", "username is required")
|
|
}
|
|
|
|
permit, err := s.acquireLoginPermit(ctx, in)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
credentialsVerified := false
|
|
defer func() {
|
|
if credentialsVerified {
|
|
permit.RecordSuccess(ctx)
|
|
return
|
|
}
|
|
permit.Release(ctx)
|
|
}()
|
|
|
|
account, err := s.accounts.GetByUsername(ctx, in.Username)
|
|
if err != nil {
|
|
permit.RecordFailure(ctx)
|
|
s.recordLoginFailure(ctx, in, "account_not_found")
|
|
return nil, response.ErrUnauthorized(40110, "invalid_credentials", "账号或密码错误")
|
|
}
|
|
|
|
if !account.IsActive() {
|
|
permit.RecordFailure(ctx)
|
|
s.recordLoginFailure(ctx, in, "account_disabled")
|
|
return nil, response.ErrForbidden(40310, "account_disabled", "账号已停用")
|
|
}
|
|
|
|
if err := bcrypt.CompareHashAndPassword([]byte(account.PasswordHash), []byte(in.Password)); err != nil {
|
|
permit.RecordFailure(ctx)
|
|
s.recordLoginFailure(ctx, in, "wrong_password")
|
|
return nil, response.ErrUnauthorized(40110, "invalid_credentials", "账号或密码错误")
|
|
}
|
|
credentialsVerified = true
|
|
|
|
token, err := s.issuer.Issue(account)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("issue token: %w", err)
|
|
}
|
|
|
|
if err := s.accounts.MarkLogin(ctx, account.ID); err != nil {
|
|
// non-fatal — login already succeeded
|
|
_ = err
|
|
}
|
|
|
|
id := account.ID
|
|
_ = s.audits.Append(ctx, domain.AuditEvent{
|
|
OperatorID: &id,
|
|
OperatorName: account.DisplayName,
|
|
Action: ActionLoginSuccess,
|
|
IP: in.IP,
|
|
UserAgent: in.UserAgent,
|
|
RequestID: in.RequestID,
|
|
})
|
|
|
|
return &LoginResult{
|
|
AccessToken: token.AccessToken,
|
|
ExpiresAt: token.ExpiresAt,
|
|
Operator: ToOperatorView(account),
|
|
}, nil
|
|
}
|
|
|
|
func (s *AuthService) acquireLoginPermit(ctx context.Context, in LoginInput) (*sharedauth.LoginPermit, error) {
|
|
if s.guard == nil {
|
|
return nil, nil
|
|
}
|
|
permit, err := s.guard.Acquire(ctx, sharedauth.LoginGuardSubject{
|
|
Identifier: in.Username,
|
|
IP: in.IP,
|
|
})
|
|
if err == nil {
|
|
return permit, nil
|
|
}
|
|
|
|
var blocked *sharedauth.LoginGuardBlockedError
|
|
if errors.As(err, &blocked) {
|
|
detail := fmt.Sprintf("请 %d 秒后重试", blocked.RetryAfterSeconds())
|
|
switch blocked.Reason {
|
|
case sharedauth.LoginGuardBlockedInProgress:
|
|
return nil, response.ErrTooManyRequests(42912, "login_in_progress", detail)
|
|
case sharedauth.LoginGuardBlockedLocked:
|
|
return nil, response.ErrTooManyRequests(42911, "login_locked", detail)
|
|
default:
|
|
return nil, response.ErrTooManyRequests(42910, "login_rate_limited", detail)
|
|
}
|
|
}
|
|
if errors.Is(err, sharedauth.ErrLoginGuardUnavailable) {
|
|
return nil, response.ErrServiceUnavailable(50305, "login_guard_unavailable", "login protection is temporarily unavailable")
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
func (s *AuthService) recordLoginFailure(ctx context.Context, in LoginInput, reason string) {
|
|
_ = s.audits.Append(ctx, domain.AuditEvent{
|
|
Action: ActionLoginFailed,
|
|
IP: in.IP,
|
|
UserAgent: in.UserAgent,
|
|
RequestID: in.RequestID,
|
|
Metadata: map[string]any{
|
|
"username": in.Username,
|
|
"reason": reason,
|
|
},
|
|
})
|
|
}
|