feat: expose signed platform health status

This commit is contained in:
Codex Local
2026-08-26 21:15:03 +08:00
parent 1e12d329dc
commit 47ef596d3b
30 changed files with 811 additions and 26 deletions
+9 -4
View File
@@ -124,6 +124,12 @@ func Run(ctx context.Context, serviceName string) error {
var alertService *alerts.Service
var diagnosticService *diagnostics.Service
var opsAccessService *opsaccess.Service
if serviceName == "api" || serviceName == "callbacks" {
diagnosticService, err = diagnostics.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository())
if err != nil {
return fmt.Errorf("create diagnostic service: %w", err)
}
}
if serviceName == "api" {
webEventService = webevents.NewService(databasePool)
cursorCodec, cursorErr := pagination.NewCodec([]byte(cfg.CursorSigningSecret))
@@ -143,10 +149,6 @@ func Run(ctx context.Context, serviceName string) error {
if err != nil {
return fmt.Errorf("create alert operations service: %w", err)
}
diagnosticService, err = diagnostics.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository())
if err != nil {
return fmt.Errorf("create diagnostic service: %w", err)
}
opsAccessService, err = opsaccess.NewService(databasePool, idempotency.NewExecutor(databasePool), audit.NewRepository(), exportStore, 5*time.Minute)
if err != nil {
return fmt.Errorf("create temporary sensitive access service: %w", err)
@@ -435,6 +437,9 @@ func Run(ctx context.Context, serviceName string) error {
ProviderCallbackSecret: []byte(cfg.ProviderCallbackSecret),
ProviderCallbackSkew: cfg.ProviderCallbackSkew,
ProviderCallbackProcessor: providerEventProcessor,
PlatformHealthCallbackSecret: []byte(cfg.PlatformHealthSecret),
PlatformHealthCallbackSkew: cfg.PlatformHealthSkew,
PlatformHealthProcessor: diagnosticService,
})
server := &http.Server{
Addr: cfg.HTTPAddr,
+1
View File
@@ -18,6 +18,7 @@ func TestCallbacksApplicationStartsAndShutsDown(t *testing.T) {
t.Setenv("REDIS_ADDR", "localhost:56379")
t.Setenv("SHUTDOWN_TIMEOUT", "1s")
t.Setenv("DEPENDENCY_TIMEOUT", "1s")
t.Setenv("PLATFORM_HEALTH_CALLBACK_SECRET", strings.Repeat("h", 32))
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
defer cancel()
if err := Run(ctx, "callbacks"); err != nil {
+16 -2
View File
@@ -59,6 +59,8 @@ type Config struct {
HeartbeatInterval int
ProviderCallbackSecret string
ProviderCallbackSkew time.Duration
PlatformHealthSecret string
PlatformHealthSkew time.Duration
MockProviderBaseURL string
MockProviderAPIKey string
VoiceProvider string
@@ -160,6 +162,10 @@ func Load(serviceName string) (Config, error) {
if err != nil {
return Config{}, err
}
platformHealthSkew, err := durationEnv("PLATFORM_HEALTH_CALLBACK_CLOCK_SKEW", 5*time.Minute)
if err != nil {
return Config{}, err
}
providerRequestTimeout, err := durationEnv("PROVIDER_REQUEST_TIMEOUT", 5*time.Second)
if err != nil {
return Config{}, err
@@ -203,7 +209,7 @@ func Load(serviceName string) (Config, error) {
environment := stringEnv("APP_ENV", EnvironmentLocal)
secretValues, err := loadSecretValues(
"DATABASE_URL", "REDIS_PASSWORD", "DEVICE_LOCAL_SECRET", "RTC_APP_KEY", "RTC_TOKEN_SIGNING_SECRET",
"CURSOR_SIGNING_SECRET", "SHARE_TOKEN_SIGNING_SECRET", "PROVIDER_CALLBACK_SECRET",
"CURSOR_SIGNING_SECRET", "SHARE_TOKEN_SIGNING_SECRET", "PROVIDER_CALLBACK_SECRET", "PLATFORM_HEALTH_CALLBACK_SECRET",
"MOCK_PROVIDER_API_KEY", "VOLC_ACCESS_KEY_ID", "VOLC_SECRET_ACCESS_KEY", "VOLC_SESSION_TOKEN",
"OBJECT_STORE_ACCESS_KEY", "OBJECT_STORE_SECRET_KEY", "ALERT_WEBHOOK_SECRET", "ARK_API_KEY", "CONTENT_SAFETY_API_KEY",
)
@@ -290,6 +296,8 @@ func Load(serviceName string) (Config, error) {
AllocationTTL: allocationTTL,
ProviderCallbackSecret: providerCallbackSecret,
ProviderCallbackSkew: providerCallbackSkew,
PlatformHealthSecret: secretValues["PLATFORM_HEALTH_CALLBACK_SECRET"],
PlatformHealthSkew: platformHealthSkew,
MockProviderBaseURL: stringEnv("MOCK_PROVIDER_BASE_URL", "http://localhost:59200"),
MockProviderAPIKey: secretDefault(secretValues["MOCK_PROVIDER_API_KEY"], "local-mock-provider-key"),
VoiceProvider: stringEnv("VOICE_PROVIDER", "mock"),
@@ -457,6 +465,9 @@ func (c Config) Validate() error {
if c.ServiceName == "callbacks" && len(c.ProviderCallbackSecret) < 32 {
return errors.New("PROVIDER_CALLBACK_SECRET must contain at least 32 characters")
}
if c.ServiceName == "callbacks" && len(c.PlatformHealthSecret) < 32 {
return errors.New("PLATFORM_HEALTH_CALLBACK_SECRET must contain at least 32 characters")
}
if c.ServiceName == "worker" {
if c.VoiceProvider != "mock" && c.VoiceProvider != "volcengine" {
return errors.New("VOICE_PROVIDER must be mock or volcengine")
@@ -470,7 +481,7 @@ func (c Config) Validate() error {
}
}
}
if c.ProviderCallbackSkew <= 0 || c.ProviderRequestTimeout <= 0 || c.ProviderFailureThreshold < 1 || c.ProviderCircuitCooldown <= 0 {
if c.ProviderCallbackSkew <= 0 || c.PlatformHealthSkew <= 0 || c.ProviderRequestTimeout <= 0 || c.ProviderFailureThreshold < 1 || c.ProviderCircuitCooldown <= 0 {
return errors.New("provider callback skew and request timeout must be positive")
}
if c.AlertWebhookTimeout <= 0 {
@@ -578,6 +589,9 @@ func (c Config) validateProduction() error {
if c.ServiceName == "callbacks" && isPlaceholderSecret(c.ProviderCallbackSecret) {
return errors.New("production PROVIDER_CALLBACK_SECRET must not use a local or example value")
}
if c.ServiceName == "callbacks" && isPlaceholderSecret(c.PlatformHealthSecret) {
return errors.New("production PLATFORM_HEALTH_CALLBACK_SECRET must not use a local or example value")
}
if c.ServiceName == "worker" {
if c.VoiceProvider != "volcengine" || len(c.RTCAppID) != 24 || isLocalValue(c.RTCAppID) || !isHTTPSURL(c.VolcRTCBaseURL) || isLocalValue(c.VolcRTCBaseURL) || isPlaceholderSecret(c.VolcAccessKeyID) || isPlaceholderSecret(c.VolcSecretAccessKey) {
return errors.New("production voice provider must use signed Volcengine RTC HTTPS with non-placeholder AK/SK")
+1
View File
@@ -10,6 +10,7 @@ import (
func TestLoadUsesServicePortDefaults(t *testing.T) {
t.Setenv("HTTP_ADDR", "")
t.Setenv("APP_ENV", "test")
t.Setenv("PLATFORM_HEALTH_CALLBACK_SECRET", strings.Repeat("h", 32))
tests := map[string]string{
"api": "0.0.0.0:58080",
+135
View File
@@ -0,0 +1,135 @@
package diagnostics
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"regexp"
"strings"
"time"
"data-terminal-cloud/internal/identity"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
const (
PlatformHealthSchema = "platform-server-health-v1"
platformHealthMaxAge = 15 * time.Minute
platformHealthFuture = 2 * time.Minute
)
var platformHealthCheckName = regexp.MustCompile(`^[a-z][a-z0-9_.-]{1,79}$`)
type PlatformHealthCheck struct {
Name string `json:"name"`
Status string `json:"status"`
Value string `json:"value"`
}
type PlatformHealthInput struct {
Schema string `json:"schema"`
Status string `json:"status"`
ObservedAt time.Time `json:"observed_at"`
Checks []PlatformHealthCheck `json:"checks"`
}
type PlatformHealthSnapshot struct {
ID string `json:"id"`
Schema string `json:"schema"`
Status string `json:"status"`
ObservedAt time.Time `json:"observed_at"`
ReceivedAt time.Time `json:"received_at"`
Checks []PlatformHealthCheck `json:"checks"`
}
type PlatformHealthReceipt struct {
ID string `json:"id"`
Status string `json:"status"`
Duplicate bool `json:"duplicate"`
}
func (s *Service) IngestPlatformHealth(ctx context.Context, input PlatformHealthInput, rawBody []byte, now time.Time) (PlatformHealthReceipt, error) {
if !validPlatformHealth(input, now) || len(rawBody) == 0 || len(rawBody) > 64*1024 {
return PlatformHealthReceipt{}, ErrInvalid
}
checks, err := json.Marshal(input.Checks)
if err != nil {
return PlatformHealthReceipt{}, err
}
digestBytes := sha256.Sum256(rawBody)
digest := hex.EncodeToString(digestBytes[:])
publicID := uuid.NewString()
var storedID, storedStatus string
err = s.pool.QueryRow(ctx, `
INSERT INTO platform_health_snapshot(public_id,schema_name,status,observed_at,checks,body_digest)
VALUES($1::uuid,$2,$3,$4,$5::jsonb,$6)
ON CONFLICT(body_digest) DO UPDATE SET body_digest=EXCLUDED.body_digest
RETURNING public_id::text,status`, publicID, input.Schema, input.Status, input.ObservedAt, checks, digest).Scan(&storedID, &storedStatus)
if err != nil {
return PlatformHealthReceipt{}, err
}
_, _ = s.pool.Exec(ctx, `DELETE FROM platform_health_snapshot WHERE received_at < now()-interval '30 days'`)
return PlatformHealthReceipt{ID: storedID, Status: storedStatus, Duplicate: storedID != publicID}, nil
}
func (s *Service) GetLatestPlatformHealth(ctx context.Context, principal identity.Principal, tenantPublicID string) (PlatformHealthSnapshot, error) {
if _, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin"); err != nil {
return PlatformHealthSnapshot{}, err
}
var item PlatformHealthSnapshot
var checks []byte
err := s.pool.QueryRow(ctx, `
SELECT public_id::text,schema_name,status,observed_at,received_at,checks
FROM platform_health_snapshot ORDER BY observed_at DESC,id DESC LIMIT 1`).Scan(
&item.ID, &item.Schema, &item.Status, &item.ObservedAt, &item.ReceivedAt, &checks,
)
if errors.Is(err, pgx.ErrNoRows) {
return PlatformHealthSnapshot{}, ErrNotFound
}
if err != nil {
return PlatformHealthSnapshot{}, err
}
if err = json.Unmarshal(checks, &item.Checks); err != nil {
return PlatformHealthSnapshot{}, err
}
return item, nil
}
func validPlatformHealth(input PlatformHealthInput, now time.Time) bool {
if input.Schema != PlatformHealthSchema || !platformHealthStatus(input.Status) || input.ObservedAt.IsZero() ||
input.ObservedAt.Before(now.Add(-platformHealthMaxAge)) || input.ObservedAt.After(now.Add(platformHealthFuture)) ||
len(input.Checks) == 0 || len(input.Checks) > 64 {
return false
}
seen := make(map[string]struct{}, len(input.Checks))
critical, warning := 0, 0
for _, check := range input.Checks {
if !platformHealthCheckName.MatchString(check.Name) || !platformHealthStatus(check.Status) || len([]rune(check.Value)) > 256 || strings.ContainsAny(check.Value, "\r\n\x00") {
return false
}
if _, exists := seen[check.Name]; exists {
return false
}
seen[check.Name] = struct{}{}
if check.Status == "critical" {
critical++
} else if check.Status == "warning" {
warning++
}
}
expected := "healthy"
if critical > 0 {
expected = "critical"
} else if warning > 0 {
expected = "warning"
}
return input.Status == expected
}
func platformHealthStatus(value string) bool {
return value == "healthy" || value == "warning" || value == "critical"
}
@@ -0,0 +1,35 @@
package diagnostics
import (
"testing"
"time"
)
func TestValidPlatformHealthRequiresConsistentAggregateAndUniqueChecks(t *testing.T) {
now := time.Now().UTC()
valid := PlatformHealthInput{
Schema: PlatformHealthSchema, Status: "warning", ObservedAt: now,
Checks: []PlatformHealthCheck{
{Name: "public_https", Status: "healthy", Value: "http_200"},
{Name: "disk_usage", Status: "warning", Value: "used_percent=81"},
},
}
if !validPlatformHealth(valid, now) {
t.Fatal("valid report rejected")
}
inconsistent := valid
inconsistent.Status = "healthy"
if validPlatformHealth(inconsistent, now) {
t.Fatal("inconsistent aggregate accepted")
}
duplicate := valid
duplicate.Checks = append(duplicate.Checks, duplicate.Checks[0])
if validPlatformHealth(duplicate, now) {
t.Fatal("duplicate check accepted")
}
stale := valid
stale.ObservedAt = now.Add(-16 * time.Minute)
if validPlatformHealth(stale, now) {
t.Fatal("stale report accepted")
}
}
+20
View File
@@ -10,6 +10,26 @@ import (
"data-terminal-cloud/internal/idempotency"
)
func GetOpsPlatformHealthHandler(service *diagnostics.Service) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
principal, ok := webPrincipal(w, r)
if !ok {
return
}
tenantID, ok := opsTenantHeader(w, r)
if !ok {
return
}
item, err := service.GetLatestPlatformHealth(r.Context(), principal, tenantID)
if err != nil {
writeDiagnosticError(w, r, err)
return
}
w.Header().Set("Cache-Control", "no-store")
WriteJSON(w, r, http.StatusOK, item)
})
}
func ListOpsIntegrationIssuesHandler(service *diagnostics.Service) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
principal, ok := webPrincipal(w, r)
@@ -0,0 +1,93 @@
package httpapi
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"strconv"
"strings"
"time"
"data-terminal-cloud/internal/diagnostics"
)
const (
HeaderPlatformTimestamp = "X-Platform-Timestamp"
HeaderPlatformSignature = "X-Platform-Signature"
)
type PlatformHealthProcessor interface {
IngestPlatformHealth(context.Context, diagnostics.PlatformHealthInput, []byte, time.Time) (diagnostics.PlatformHealthReceipt, error)
}
func PlatformHealthCallbackHandler(secret []byte, maxSkew time.Duration, processor PlatformHealthProcessor) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
r.Body = http.MaxBytesReader(w, r.Body, 64*1024)
rawBody, err := io.ReadAll(r.Body)
if err != nil {
WriteError(w, r, http.StatusBadRequest, "INVALID_PLATFORM_HEALTH_REPORT", "服务器健康报告无法读取")
return
}
now := time.Now().UTC()
if !verifyPlatformHealthSignature(secret, r.Header.Get(HeaderPlatformTimestamp), r.Header.Get(HeaderPlatformSignature), rawBody, now, maxSkew) {
WriteError(w, r, http.StatusUnauthorized, "PLATFORM_HEALTH_AUTHENTICATION_FAILED", "服务器健康报告身份验证失败")
return
}
decoder := json.NewDecoder(bytes.NewReader(rawBody))
decoder.DisallowUnknownFields()
var input diagnostics.PlatformHealthInput
if err = decoder.Decode(&input); err != nil || ensureJSONEOF(decoder) != nil {
WriteError(w, r, http.StatusBadRequest, "INVALID_PLATFORM_HEALTH_REPORT", "服务器健康报告字段不正确")
return
}
receipt, err := processor.IngestPlatformHealth(r.Context(), input, rawBody, now)
if err != nil {
if errors.Is(err, diagnostics.ErrInvalid) {
WriteError(w, r, http.StatusBadRequest, "INVALID_PLATFORM_HEALTH_REPORT", "服务器健康报告字段不正确")
return
}
slog.ErrorContext(r.Context(), "platform health callback processing failed", "error", err)
WriteAPIError(w, r, http.StatusServiceUnavailable, Error{Code: "PLATFORM_HEALTH_UNAVAILABLE", Message: "服务器健康报告暂时无法处理", Retryable: true})
return
}
status := http.StatusCreated
if receipt.Duplicate {
status = http.StatusOK
}
w.Header().Set("Cache-Control", "no-store")
WriteJSON(w, r, status, receipt)
})
}
func verifyPlatformHealthSignature(secret []byte, timestamp, signature string, body []byte, now time.Time, maxSkew time.Duration) bool {
if len(secret) < 32 || maxSkew <= 0 || timestamp == "" || !strings.HasPrefix(signature, "sha256=") {
return false
}
epoch, err := strconv.ParseInt(timestamp, 10, 64)
if err != nil || absDuration(now.Sub(time.Unix(epoch, 0))) > maxSkew {
return false
}
provided, err := hex.DecodeString(strings.TrimPrefix(signature, "sha256="))
if err != nil {
return false
}
mac := hmac.New(sha256.New, secret)
_, _ = mac.Write([]byte(timestamp))
_, _ = mac.Write([]byte("."))
_, _ = mac.Write(body)
return hmac.Equal(provided, mac.Sum(nil))
}
func absDuration(value time.Duration) time.Duration {
if value < 0 {
return -value
}
return value
}
@@ -0,0 +1,78 @@
package httpapi
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"testing"
"time"
"data-terminal-cloud/internal/diagnostics"
)
type fakePlatformHealthProcessor struct {
calls int
receipt diagnostics.PlatformHealthReceipt
err error
lastInput diagnostics.PlatformHealthInput
}
func (p *fakePlatformHealthProcessor) IngestPlatformHealth(_ context.Context, input diagnostics.PlatformHealthInput, _ []byte, _ time.Time) (diagnostics.PlatformHealthReceipt, error) {
p.calls++
p.lastInput = input
return p.receipt, p.err
}
func TestPlatformHealthCallbackRequiresSignatureAndAcceptsReport(t *testing.T) {
secret := []byte("01234567890123456789012345678901")
now := time.Now().UTC().Truncate(time.Second)
body, _ := json.Marshal(diagnostics.PlatformHealthInput{
Schema: diagnostics.PlatformHealthSchema, Status: "healthy", ObservedAt: now,
Checks: []diagnostics.PlatformHealthCheck{{Name: "public_https", Status: "healthy", Value: "http_200"}},
})
processor := &fakePlatformHealthProcessor{receipt: diagnostics.PlatformHealthReceipt{ID: "10000000-0000-0000-0000-000000000999", Status: "healthy"}}
handler := PlatformHealthCallbackHandler(secret, time.Minute, processor)
unsigned := httptest.NewRequest(http.MethodPost, "/callbacks/v1/platform-health", strings.NewReader(string(body)))
unsignedRecorder := httptest.NewRecorder()
handler.ServeHTTP(unsignedRecorder, unsigned)
if unsignedRecorder.Code != http.StatusUnauthorized || processor.calls != 0 {
t.Fatalf("unsigned status=%d calls=%d", unsignedRecorder.Code, processor.calls)
}
timestamp := strconv.FormatInt(now.Unix(), 10)
signed := httptest.NewRequest(http.MethodPost, "/callbacks/v1/platform-health", strings.NewReader(string(body)))
signed.Header.Set(HeaderPlatformTimestamp, timestamp)
signed.Header.Set(HeaderPlatformSignature, signPlatformHealth(secret, timestamp, body))
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, signed)
if recorder.Code != http.StatusCreated || processor.calls != 1 || processor.lastInput.Schema != diagnostics.PlatformHealthSchema {
t.Fatalf("status=%d calls=%d input=%+v body=%s", recorder.Code, processor.calls, processor.lastInput, recorder.Body.String())
}
}
func TestPlatformHealthCallbackRejectsStaleAndMalformedSignature(t *testing.T) {
secret := []byte("01234567890123456789012345678901")
body := []byte(`{"schema":"platform-server-health-v1","status":"healthy","observed_at":"2026-08-26T00:00:00Z","checks":[]}`)
now := time.Now().UTC()
staleTimestamp := strconv.FormatInt(now.Add(-10*time.Minute).Unix(), 10)
if verifyPlatformHealthSignature(secret, staleTimestamp, signPlatformHealth(secret, staleTimestamp, body), body, now, time.Minute) {
t.Fatal("stale signature accepted")
}
if verifyPlatformHealthSignature(secret, strconv.FormatInt(now.Unix(), 10), "sha256=broken", body, now, time.Minute) {
t.Fatal("malformed signature accepted")
}
}
func signPlatformHealth(secret []byte, timestamp string, body []byte) string {
mac := hmac.New(sha256.New, secret)
_, _ = mac.Write([]byte(timestamp + "."))
_, _ = mac.Write(body)
return "sha256=" + hex.EncodeToString(mac.Sum(nil))
}
+9
View File
@@ -48,6 +48,9 @@ type RouterOptions struct {
ProviderCallbackSecret []byte
ProviderCallbackSkew time.Duration
ProviderCallbackProcessor ProviderCallbackProcessor
PlatformHealthCallbackSecret []byte
PlatformHealthCallbackSkew time.Duration
PlatformHealthProcessor PlatformHealthProcessor
}
func NewRouter(options RouterOptions) http.Handler {
@@ -117,6 +120,7 @@ func NewRouter(options RouterOptions) http.Handler {
mux.Handle("GET /api/v1/ops/cost-summary", authenticated(GetOpsCostSummaryHandler(options.BillingService)))
}
if options.DiagnosticService != nil {
mux.Handle("GET /api/v1/ops/diagnostics/platform-health", authenticated(GetOpsPlatformHealthHandler(options.DiagnosticService)))
mux.Handle("GET /api/v1/ops/integration-issues", authenticated(ListOpsIntegrationIssuesHandler(options.DiagnosticService)))
mux.Handle("POST /api/v1/ops/integration-issues", authenticated(CreateOpsIntegrationIssueHandler(options.DiagnosticService)))
mux.Handle("PATCH /api/v1/ops/integration-issues/{id}", authenticated(UpdateOpsIntegrationIssueHandler(options.DiagnosticService)))
@@ -278,6 +282,11 @@ func NewRouter(options RouterOptions) http.Handler {
options.ProviderCallbackSecret, options.ProviderCallbackSkew, options.ProviderCallbackProcessor,
))
}
if options.ServiceName == "callbacks" && len(options.PlatformHealthCallbackSecret) > 0 && options.PlatformHealthProcessor != nil {
mux.Handle("POST /callbacks/v1/platform-health", PlatformHealthCallbackHandler(
options.PlatformHealthCallbackSecret, options.PlatformHealthCallbackSkew, options.PlatformHealthProcessor,
))
}
mux.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
WriteError(writer, request, http.StatusNotFound, "RESOURCE_NOT_FOUND", "请求的资源不存在")
})