chii
eb972dfa93
单二进制三角色 + Dock 适配器 + Swarm stack 达到可部署态；mgr1 实测订阅经 central-proxy bootstrap，探活 alive=41/52。 Co-authored-by: Cursor <cursoragent@cursor.com>
304 lines
9.7 KiB
Go
304 lines
9.7 KiB
Go
// server.go：gateway 依赖容器与 HTTP 路由（/v1 /mcp /admin /healthz /readyz）。
|
||
package gateway
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"onesvm.com/onesvm/browser-server/internal/auth"
|
||
"onesvm.com/onesvm/browser-server/internal/contract"
|
||
"onesvm.com/onesvm/browser-server/internal/policy"
|
||
"onesvm.com/onesvm/browser-server/internal/store"
|
||
)
|
||
|
||
// Deps gateway 依赖容器（main.go 组装注入；Port-Adapter 纪律：编排不摸表）。
|
||
type Deps struct {
|
||
DB *store.DB
|
||
Glue *DBGlue // store 扩展查询（KeysAll/ConsumerByName）
|
||
Verifier KeyVerifier
|
||
Policy *policy.Engine
|
||
Scheduler *SchedulerClient
|
||
Cache *SearchCache
|
||
Limiter *RateLimiter
|
||
AdminToken string // X-Service-Token 管理面比对值（MustEnv 注入）
|
||
SeedSalt string // admin 签发同源盐
|
||
Logger *log.Logger
|
||
}
|
||
|
||
// KeyVerifier 认证/签发接口（auth.Verifier 满足；镜像 auth.Error 形状解耦实现）。
|
||
type KeyVerifier interface {
|
||
Check(plaintext string) (*store.ApiKey, *auth.Error)
|
||
ReserveDaily(keyID int64, dailyLimit int, now time.Time) (int, *auth.Error)
|
||
Release(keyID int64, now time.Time) error
|
||
Settle(keyID int64, now time.Time) error
|
||
Issue(consumerID int64, name string, scopes []string, rpm, daily, monthly, sessions int, expiresAt *time.Time) (string, int64, error)
|
||
Salt() string
|
||
}
|
||
|
||
// Server gateway HTTP 服务。
|
||
type Server struct {
|
||
deps Deps
|
||
// inFlight 在途会话计数（按 key，X-Session-Remaining 头）
|
||
mu sync.Mutex
|
||
inFlight map[int64]int
|
||
// revokeCache 吊销收敛缓存：keyID→吊销时间（Check 由 store 直查，本 map 预留）
|
||
lastReady bool
|
||
}
|
||
|
||
// NewServer 构造。
|
||
func NewServer(d Deps) *Server {
|
||
return &Server{deps: d, inFlight: map[int64]int{}}
|
||
}
|
||
|
||
// Handler 组装路由。
|
||
func (s *Server) Handler() http.Handler {
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("GET /healthz", s.handleHealthz)
|
||
mux.HandleFunc("GET /readyz", s.handleReadyz)
|
||
mux.HandleFunc("POST /v1/search", s.consumer(s.handleV1Search))
|
||
mux.HandleFunc("POST /v1/read", s.consumer(s.handleV1Read))
|
||
// ITER-1 F3（fail-w5-smoke-iter1）：MCP 2026 无状态分级鉴权——
|
||
// initialize/tools/list/ping 免 key（连接协商面）；tools/call 保留 consumer
|
||
// 鉴权（mcp-usage §1.2 消费者契约是「调用工具须 key」，协商面无 key 不冲突）。
|
||
mux.HandleFunc("POST /mcp", s.mcpMux)
|
||
mux.HandleFunc("POST /admin/keys", s.admin(s.handleAdminIssue))
|
||
mux.HandleFunc("DELETE /admin/keys/{id}", s.admin(s.handleAdminRevoke))
|
||
mux.HandleFunc("GET /admin/keys", s.admin(s.handleAdminList))
|
||
return mux
|
||
}
|
||
|
||
// mcpAuthFree MCP 免鉴权方法集（连接协商 + 通知；2026 无状态规范）。
|
||
var mcpAuthFree = map[string]bool{"initialize": true, "tools/list": true, "ping": true, "notifications/initialized": true}
|
||
|
||
// mcpMux MCP 按方法分级：免鉴权方法直通 handleMCP（a=nil），
|
||
// 其余（tools/call 等）走 consumer 鉴权（handler 内校验在工具层）。
|
||
// 探测读取的 body 原样回注（GetBody + 替换 Body），下游 handler 仍按完整 body 解析。
|
||
func (s *Server) mcpMux(w http.ResponseWriter, r *http.Request) {
|
||
body, err := io.ReadAll(io.LimitReader(r.Body, 4*1024*1024))
|
||
_ = r.Body.Close()
|
||
if err != nil {
|
||
s.writeErr(w, r, contract.CodeUnauthorized, "请求体读取失败", http.StatusBadRequest, "", nil)
|
||
return
|
||
}
|
||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||
var probe struct {
|
||
Method string `json:"method"`
|
||
}
|
||
if json.Unmarshal(body, &probe) == nil && mcpAuthFree[probe.Method] {
|
||
s.handleMCP(w, r, nil)
|
||
return
|
||
}
|
||
s.consumer(s.handleMCP)(w, r)
|
||
}
|
||
|
||
// ---------- 中间件 ----------
|
||
|
||
// consumerAuth 认证上下文。
|
||
type consumerAuth struct {
|
||
Key *store.ApiKey
|
||
Remaining int // 日配额剩余（预扣后）
|
||
}
|
||
|
||
// ctxKeyAuth 上下文键。
|
||
type ctxKeyAuth struct{}
|
||
|
||
// consumer consumer 认证中间件（X-Service-Token，禁 Bearer）。
|
||
func (s *Server) consumer(next func(http.ResponseWriter, *http.Request, *consumerAuth)) http.HandlerFunc {
|
||
return func(w http.ResponseWriter, r *http.Request) {
|
||
tok := r.Header.Get("X-Service-Token")
|
||
if tok == "" && strings.HasPrefix(r.Header.Get("Authorization"), "Bearer ") {
|
||
w.Header().Set("WWW-Authenticate", `X-Service-Token realm="browser-server", error="invalid_token", hint="禁止 Authorization: Bearer，改用 X-Service-Token"`)
|
||
}
|
||
if tok == "" {
|
||
s.writeErr(w, r, contract.CodeUnauthorized, "key 缺失（头 X-Service-Token）", http.StatusUnauthorized, "", nil)
|
||
return
|
||
}
|
||
key, aerr := s.deps.Verifier.Check(tok)
|
||
if aerr != nil {
|
||
s.writeErr(w, r, aerr.Code, aerr.Message, http.StatusUnauthorized, "", nil)
|
||
return
|
||
}
|
||
next(w, r, &consumerAuth{Key: key})
|
||
}
|
||
}
|
||
|
||
// admin admin 认证中间件（X-Service-Token == admin token；hmac 比对）。
|
||
func (s *Server) admin(next http.HandlerFunc) http.HandlerFunc {
|
||
return func(w http.ResponseWriter, r *http.Request) {
|
||
tok := r.Header.Get("X-Service-Token")
|
||
if !s.adminTokenOK(tok) {
|
||
s.writeErr(w, r, contract.CodeUnauthorized, "admin token 无效", http.StatusUnauthorized, "", nil)
|
||
return
|
||
}
|
||
next(w, r)
|
||
}
|
||
}
|
||
|
||
// adminTokenOK 常数时间比对（T3 纪律）。
|
||
func (s *Server) adminTokenOK(tok string) bool {
|
||
return constTimeEqual(tok, s.deps.AdminToken)
|
||
}
|
||
|
||
// constTimeEqual 长度归一的常数时间比对。
|
||
func constTimeEqual(a, b string) bool {
|
||
if len(a) != len(b) {
|
||
// 长度不同仍跑一遍比对抹平时序差
|
||
_ = hmacEqual([]byte(a), make([]byte, len(a)))
|
||
return false
|
||
}
|
||
return hmacEqual([]byte(a), []byte(b))
|
||
}
|
||
|
||
// ---------- /healthz /readyz ----------
|
||
|
||
func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
_, _ = w.Write([]byte(`{"ok":true}`))
|
||
}
|
||
|
||
// handleReadyz 依赖 scheduler /pressure 可达性；不可达 500。
|
||
func (s *Server) handleReadyz(w http.ResponseWriter, r *http.Request) {
|
||
if err := s.deps.Scheduler.Pressure(r.Context()); err != nil {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(http.StatusInternalServerError)
|
||
_ = json.NewEncoder(w).Encode(map[string]any{"ok": false, "error": err.Error()})
|
||
return
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
_, _ = w.Write([]byte(`{"ok":true}`))
|
||
}
|
||
|
||
// ---------- 统一响应 ----------
|
||
|
||
// writeErr 统一错误信封输出：HTTP 状态与 error.code 按 mcp-usage §3 映射。
|
||
// 200 信封类错误（blocked/extract_failed/upstream/timeout 经 pipeline 归一时）由
|
||
// pipeline 直接产出完整信封；本函数仅服务请求前置错误（401/402/403/429/503）。
|
||
func (s *Server) writeErr(w http.ResponseWriter, r *http.Request, code, msg string, status int, ruleID string, retryAfter *int) {
|
||
env := contract.Envelope{
|
||
OK: false,
|
||
Kind: kindOfPath(r.URL.Path),
|
||
RequestID: requestIDOf(r),
|
||
Error: &contract.ErrBody{Code: code, Message: msg, RetryAfterS: retryAfter},
|
||
}
|
||
if ruleID != "" {
|
||
env.Error.Message = msg + "（rule_id=" + ruleID + "）"
|
||
}
|
||
w.Header().Set("Content-Type", "application/json")
|
||
if retryAfter != nil {
|
||
w.Header().Set("Retry-After", strconv.Itoa(*retryAfter))
|
||
}
|
||
w.WriteHeader(status)
|
||
_ = json.NewEncoder(w).Encode(env)
|
||
}
|
||
|
||
// kindOfPath 由路径推断信封 kind（错误信封字段裁剪用）。
|
||
func kindOfPath(p string) string {
|
||
switch {
|
||
case strings.HasSuffix(p, "/search"):
|
||
return "search"
|
||
case strings.HasSuffix(p, "/read"):
|
||
return "read"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// requestIDOf 请求级 ID（无真实请求 ID 前，错误信封给空串——信封字段纪律：恒存在）。
|
||
func requestIDOf(_ *http.Request) string { return "" }
|
||
|
||
// ---------- 在途会话计数（X-Session-Remaining） ----------
|
||
|
||
// beginSession 在途 +1；返回剩余额度。
|
||
func (s *Server) beginSession(keyID int64, concurrentLimit int) int {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
s.inFlight[keyID]++
|
||
if concurrentLimit <= 0 {
|
||
return 0
|
||
}
|
||
rem := concurrentLimit - s.inFlight[keyID]
|
||
if rem < 0 {
|
||
rem = 0
|
||
}
|
||
return rem
|
||
}
|
||
|
||
// endSession 在途 -1。
|
||
func (s *Server) endSession(keyID int64) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if n, ok := s.inFlight[keyID]; ok {
|
||
if n <= 1 {
|
||
delete(s.inFlight, keyID)
|
||
} else {
|
||
s.inFlight[keyID] = n - 1
|
||
}
|
||
}
|
||
}
|
||
|
||
// sessionRemaining 当前剩余（不增减）。
|
||
func (s *Server) sessionRemaining(keyID int64, concurrentLimit int) int {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if concurrentLimit <= 0 {
|
||
return 0
|
||
}
|
||
rem := concurrentLimit - s.inFlight[keyID]
|
||
if rem < 0 {
|
||
rem = 0
|
||
}
|
||
return rem
|
||
}
|
||
|
||
// ---------- 请求处理上下文 ----------
|
||
|
||
// runCtx 单请求管线上下文（pipeline.Run 输入）。
|
||
type runCtx struct {
|
||
Intent string
|
||
Search *contract.SearchInput
|
||
Read *contract.ReadInput
|
||
Auth *consumerAuth
|
||
W http.ResponseWriter
|
||
R *http.Request
|
||
}
|
||
|
||
// beginRequest 请求前置于途计数。
|
||
func (s *Server) beginRequest(a *consumerAuth) {
|
||
a.Remaining = s.beginSession(a.Key.ID, a.Key.ConcurrentSessions)
|
||
}
|
||
|
||
// finishRequest 请求结束后释放在途计数。
|
||
func (s *Server) finishRequest(a *consumerAuth) {
|
||
s.endSession(a.Key.ID)
|
||
}
|
||
|
||
// writeEnvelope 成功/信封错误统一输出（带 X-Session-Remaining）。
|
||
func (s *Server) writeEnvelope(w http.ResponseWriter, r *http.Request, a *consumerAuth, body []byte) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
if a != nil {
|
||
rem := s.sessionRemaining(a.Key.ID, a.Key.ConcurrentSessions)
|
||
w.Header().Set("X-Session-Remaining", strconv.Itoa(rem))
|
||
}
|
||
w.WriteHeader(http.StatusOK)
|
||
_, _ = w.Write(body)
|
||
}
|
||
|
||
// jsonDecode 严格单对象 JSON 解码。
|
||
func jsonDecode(r *http.Request, v any) error {
|
||
dec := json.NewDecoder(r.Body)
|
||
if err := dec.Decode(v); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ctxOf 请求 ctx（pipeline 传参用）。
|
||
func ctxOf(r *http.Request) context.Context { return r.Context() }
|