chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(admin): add atomic task claim leases

Browse Source
This commit is contained in:
QiuSW
2026-08-04 22:33:16 +08:00
parent 526af1eb31
commit 5f060f60ee
20 changed files with 2923 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin/internal/server/evidence_test.go
+27
View File
@@ -134,6 +134,7 @@ func TestRealDeviceCredentialIdentityIsolationAndMixedCredentials(t *testing.T)
if err != nil {
t.Fatalf("issue credential: %v", err)
}
insertEvidenceClaim(t, database, issued.DeviceID)
authenticator, err := deviceauth.NewSQLiteAuthenticator(database)
if err != nil {
t.Fatalf("new authenticator: %v", err)
@@ -353,6 +354,8 @@ func newEvidenceRouter(t *testing.T, authenticator deviceauth.Authenticator) (ht
t.Fatalf("migrate database: %v", err)
}
insertEvidenceAttempt(t, database)
insertEvidenceClaimDevice(t, database, evidenceDeviceID)
insertEvidenceClaim(t, database, evidenceDeviceID)
store, err := evidencestorage.NewStore(database, filepath.Join(t.TempDir(), "assets"))
if err != nil {
t.Fatalf("new evidence store: %v", err)
@@ -384,6 +387,30 @@ func insertEvidenceAttempt(t *testing.T, database *sql.DB) {
}
}
func insertEvidenceClaimDevice(t *testing.T, database *sql.DB, deviceID string) {
t.Helper()
digest := sha256.Sum256([]byte("fake evidence device"))
if _, err := database.Exec(`INSERT INTO device_credentials
(device_id,display_name,token_sha256,status,created_at,revoked_at)
VALUES (?, 'fake evidence device', ?, 'ACTIVE', '2026-08-04T00:00:00Z', NULL)`, deviceID, digest[:]); err != nil {
t.Fatalf("insert evidence device: %v", err)
}
}
func insertEvidenceClaim(t *testing.T, database *sql.DB, deviceID string) {
t.Helper()
if _, err := database.Exec(`INSERT INTO purchase_attempt_claims
(attempt_id,task_id,authorization_id,claimed_by_device_id,session_id,claim_generation,
task_version,task_title,authorization_task_version,goods_id,sku_color,sku_size,quantity,
total_price_cap,authorization_expires_at,claim_nonce,claim_token_sha256,lease_expires_at,claimed_at,closed_at)
VALUES (?, ?, ?, ?, '23c9f507-7473-4fa6-8d71-8786c34c6301', 1, 1, 'task',
1, '123', 'black', 'M', 1, '1.00', '2026-08-04T00:00:00Z', ?, ?,
'2026-08-04T02:00:00Z', '2026-08-04T00:00:00Z', NULL)`, evidenceAttemptID,
evidenceTaskID, evidenceAuthID, deviceID, bytes.Repeat([]byte{1}, 32), bytes.Repeat([]byte{2}, 32)); err != nil {
t.Fatalf("insert evidence claim: %v", err)
}
}
func serveEvidenceUpload(t *testing.T, router http.Handler, taskID string, fields map[string]string, file []byte, fileContentType, filename string, extra func(*multipart.Writer) error) *httptest.ResponseRecorder {
t.Helper()
request := newEvidenceUploadRequest(t, taskID, fields, file, fileContentType, filename, extra)
admin/internal/server/router.go
+5 -1
View File
@@ -16,6 +16,7 @@ import (
"cmbuyer/admin/internal/auth"
"cmbuyer/admin/internal/deviceauth"
"cmbuyer/admin/internal/evidence"
"cmbuyer/admin/internal/taskclaim"
"cmbuyer/admin/internal/taskdetail"
"cmbuyer/admin/internal/tasks"
"cmbuyer/admin/internal/transport/webui"
@@ -36,11 +37,12 @@ type Options struct {
TaskDetails taskdetail.Store
Evidence evidence.Store
DeviceAuthenticator deviceauth.Authenticator
TaskClaims taskclaim.Service
}
// NewRouter 返回当前服务范围内的完整 HTTP 路由。
func NewRouter(options Options) (*gin.Engine, error) {
if options.AdminUsername == "" || options.AdminPasswordBcrypt == "" || options.Sessions == nil || options.Tasks == nil || options.TaskDetails == nil || options.Evidence == nil || options.DeviceAuthenticator == nil {
if options.AdminUsername == "" || options.AdminPasswordBcrypt == "" || options.Sessions == nil || options.Tasks == nil || options.TaskDetails == nil || options.Evidence == nil || options.DeviceAuthenticator == nil || options.TaskClaims == nil {
return nil, errors.New("server authentication options are incomplete")
}
@@ -57,6 +59,8 @@ func NewRouter(options Options) (*gin.Engine, error) {
router.POST("/tasks", createTask(options))
router.POST("/tasks/start-purchases", startPurchases(options))
router.POST("/api/v1/tasks/:id/evidence", uploadEvidence(options))
router.POST("/api/v1/tasks/claim-next", claimNext(options))
router.POST("/api/v1/tasks/:id/lease/renew", renewLease(options))
router.GET("/evidence/:asset_id", readEvidence(options))
router.GET("/static/tasks.js", func(context *gin.Context) {
context.Data(http.StatusOK, "application/javascript; charset=utf-8", webui.TasksScript())
admin/internal/server/router_test.go
+16
View File
@@ -15,6 +15,7 @@ import (
"cmbuyer/admin/internal/deviceauth"
"cmbuyer/admin/internal/evidence"
"cmbuyer/admin/internal/server"
"cmbuyer/admin/internal/taskclaim"
"cmbuyer/admin/internal/taskdetail"
"cmbuyer/admin/internal/tasks"
@@ -493,6 +494,10 @@ func newRouterWithStore(t *testing.T, store tasks.Store) (*gin.Engine, *auth.Man
}
func newRouterWithDependencies(t *testing.T, store tasks.Store, details taskdetail.Store, evidenceStore evidence.Store, deviceAuthenticator deviceauth.Authenticator) (*gin.Engine, *auth.Manager) {
return newRouterWithClaimService(t, store, details, evidenceStore, deviceAuthenticator, emptyTaskClaimService{})
}
func newRouterWithClaimService(t *testing.T, store tasks.Store, details taskdetail.Store, evidenceStore evidence.Store, deviceAuthenticator deviceauth.Authenticator, claims taskclaim.Service) (*gin.Engine, *auth.Manager) {
t.Helper()
gin.SetMode(gin.TestMode)
hash, err := bcrypt.GenerateFromPassword([]byte("test-password"), bcrypt.MinCost)
@@ -508,6 +513,7 @@ func newRouterWithDependencies(t *testing.T, store tasks.Store, details taskdeta
TaskDetails: details,
Evidence: evidenceStore,
DeviceAuthenticator: deviceAuthenticator,
TaskClaims: claims,
})
if err != nil {
t.Fatalf("NewRouter: %v", err)
@@ -517,6 +523,16 @@ func newRouterWithDependencies(t *testing.T, store tasks.Store, details taskdeta
type emptyDetailStore struct{}
type emptyTaskClaimService struct{}
func (emptyTaskClaimService) ClaimNext(context.Context, string, taskclaim.ClaimCommand) (taskclaim.ClaimResponse, bool, error) {
return taskclaim.ClaimResponse{}, false, nil
}
func (emptyTaskClaimService) Renew(context.Context, string, taskclaim.RenewCommand) (taskclaim.RenewResponse, error) {
return taskclaim.RenewResponse{}, taskclaim.ErrNotCurrent
}
func (emptyDetailStore) Get(context.Context, string) (taskdetail.Detail, error) {
return taskdetail.Detail{}, taskdetail.ErrNotFound
}
admin/internal/server/task_claims.go
+164
View File
@@ -0,0 +1,164 @@
package server
import (
"bytes"
"encoding/json"
"errors"
"io"
"net/http"
"unicode/utf8"
"cmbuyer/admin/internal/deviceauth"
"cmbuyer/admin/internal/taskclaim"
"github.com/gin-gonic/gin"
)
const maxClaimJSONBytes = 4096
func claimNext(options Options) gin.HandlerFunc {
return func(context *gin.Context) {
principal, ok := authenticateDevice(context, options)
if !ok {
return
}
var command taskclaim.ClaimCommand
if !decodeClaimJSON(context, &command) {
return
}
response, found, err := options.TaskClaims.ClaimNext(context.Request.Context(), principal.ID, command)
if err != nil {
writeTaskClaimError(context, err)
return
}
if !found {
context.Status(http.StatusNoContent)
return
}
context.JSON(http.StatusOK, response)
}
}
func renewLease(options Options) gin.HandlerFunc {
return func(context *gin.Context) {
principal, ok := authenticateDevice(context, options)
if !ok {
return
}
var command taskclaim.RenewCommand
if !decodeClaimJSON(context, &command) {
return
}
command.TaskID = context.Param("id")
response, err := options.TaskClaims.Renew(context.Request.Context(), principal.ID, command)
if err != nil {
writeTaskClaimError(context, err)
return
}
context.JSON(http.StatusOK, response)
}
}
// Authentication precedes path interpretation, Content-Type parsing and every body read. This
// keeps rejected devices from using parsing differences as an oracle or making the server buffer data.
func authenticateDevice(context *gin.Context, options Options) (deviceauth.Principal, bool) {
principal, err := options.DeviceAuthenticator.Authenticate(context.Request)
if errors.Is(err, deviceauth.ErrUnauthenticated) {
context.Header("WWW-Authenticate", "Bearer")
context.Status(http.StatusUnauthorized)
return deviceauth.Principal{}, false
}
if err != nil || !deviceauth.ValidDeviceID(principal.ID) {
context.Status(http.StatusServiceUnavailable)
return deviceauth.Principal{}, false
}
return principal, true
}
func decodeClaimJSON(context *gin.Context, target any) bool {
if !isJSONContentType(context.GetHeader("Content-Type")) {
writeFixedError(context, http.StatusUnsupportedMediaType, "unsupported_media_type")
return false
}
context.Request.Body = http.MaxBytesReader(context.Writer, context.Request.Body, maxClaimJSONBytes)
raw, err := io.ReadAll(context.Request.Body)
if err != nil {
var tooLarge *http.MaxBytesError
if errors.As(err, &tooLarge) {
writeFixedError(context, http.StatusRequestEntityTooLarge, "request_too_large")
} else {
writeFixedError(context, http.StatusBadRequest, "invalid_request")
}
return false
}
if len(raw) == 0 || !utf8.Valid(raw) {
writeFixedError(context, http.StatusBadRequest, "invalid_request")
return false
}
if !hasUniqueTopLevelJSONFields(raw) {
writeFixedError(context, http.StatusBadRequest, "invalid_request")
return false
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.DisallowUnknownFields()
if err := decoder.Decode(target); err != nil {
writeFixedError(context, http.StatusBadRequest, "invalid_request")
return false
}
var extra any
if err := decoder.Decode(&extra); err != io.EOF {
writeFixedError(context, http.StatusBadRequest, "invalid_request")
return false
}
return true
}
func hasUniqueTopLevelJSONFields(raw []byte) bool {
decoder := json.NewDecoder(bytes.NewReader(raw))
first, err := decoder.Token()
if err != nil || first != json.Delim('{') {
return false
}
seen := make(map[string]struct{})
for decoder.More() {
key, err := decoder.Token()
name, ok := key.(string)
if err != nil || !ok {
return false
}
if _, duplicate := seen[name]; duplicate {
return false
}
seen[name] = struct{}{}
var value json.RawMessage
if err := decoder.Decode(&value); err != nil {
return false
}
}
last, err := decoder.Token()
return err == nil && last == json.Delim('}')
}
func writeTaskClaimError(context *gin.Context, err error) {
switch {
case errors.Is(err, taskclaim.ErrInvalid):
writeFixedError(context, http.StatusBadRequest, "invalid_request")
case errors.Is(err, taskclaim.ErrIdempotencyConflict):
writeFixedError(context, http.StatusConflict, "idempotency_conflict")
case errors.Is(err, taskclaim.ErrRequiresManual):
writeFixedError(context, http.StatusConflict, "claim_requires_manual")
case errors.Is(err, taskclaim.ErrNotCurrent):
writeFixedError(context, http.StatusConflict, "claim_not_current")
case errors.Is(err, taskclaim.ErrDeviceInactive):
context.Header("WWW-Authenticate", "Bearer")
context.Status(http.StatusUnauthorized)
default:
// Storage and transaction failures are intentionally bodyless: SQL, paths and candidate
// details are server-only and must not become a device-facing diagnostic oracle.
context.Status(http.StatusServiceUnavailable)
}
}
func writeFixedError(context *gin.Context, status int, code string) {
context.JSON(status, gin.H{"error": code})
}
admin/internal/server/task_claims_test.go
+166
View File
@@ -0,0 +1,166 @@
package server_test
import (
"bytes"
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"cmbuyer/admin/internal/deviceauth"
"cmbuyer/admin/internal/taskclaim"
)
const (
claimDeviceID = "10000000-0000-4000-8000-000000000001"
claimSessionID = "20000000-0000-4000-8000-000000000001"
claimRequestID = "30000000-0000-4000-8000-000000000001"
claimTaskID = "40000000-0000-4000-8000-000000000001"
claimAttemptID = "50000000-0000-4000-8000-000000000001"
claimRenewID = "60000000-0000-4000-8000-000000000001"
)
func TestTaskClaimEndpointsAuthenticateBeforeBody(t *testing.T) {
for _, authentication := range []struct {
name string
err error
status int
}{
{"unauthenticated", deviceauth.ErrUnauthenticated, http.StatusUnauthorized},
{"authentication storage unavailable", deviceauth.ErrUnavailable, http.StatusServiceUnavailable},
} {
t.Run(authentication.name, func(t *testing.T) {
authenticator := &fakeDeviceAuthenticator{err: authentication.err}
service := &fakeTaskClaimService{}
router, _ := newRouterWithClaimService(t, &memoryStore{}, emptyDetailStore{}, emptyEvidenceStore{}, authenticator, service)
for _, path := range []string{"/api/v1/tasks/claim-next", "/api/v1/tasks/" + claimTaskID + "/lease/renew"} {
body := &poisonBody{}
request := httptest.NewRequest(http.MethodPost, path, nil)
request.Body = body
response := httptest.NewRecorder()
router.ServeHTTP(response, request)
if response.Code != authentication.status || response.Body.Len() != 0 || body.reads != 0 || service.calls != 0 {
t.Fatalf("%s = status %d, body %q, reads %d, calls %d", path, response.Code, response.Body.String(), body.reads, service.calls)
}
}
})
}
}
func TestClaimNextStrictJSONSuccessEmptyAndErrors(t *testing.T) {
authenticator := &fakeDeviceAuthenticator{principal: deviceauth.Principal{ID: claimDeviceID}}
service := &fakeTaskClaimService{claimResponse: taskclaim.ClaimResponse{
Task: taskclaim.ClaimedTask{ID: claimTaskID, Version: 3, Title: "测试", ProductURL: "https://mobile.yangkeduo.com/goods.html?goods_id=1", GoodsID: "1", SKUColor: "黑色", SKUSize: "M", Quantity: 1, MaxTotalPrice: "1.00"},
Authorization: taskclaim.ClaimedAuthorization{ID: "70000000-0000-4000-8000-000000000001", TaskVersion: 2, ExpiresAt: "2026-08-04T01:10:00Z"},
Attempt: taskclaim.ClaimedAttempt{ID: claimAttemptID, ClaimToken: strings.Repeat("a", 64), ClaimGeneration: 1, LeaseExpiresAt: "2026-08-04T01:03:00Z"},
}, claimFound: true}
router, _ := newRouterWithClaimService(t, &memoryStore{}, emptyDetailStore{}, emptyEvidenceStore{}, authenticator, service)
valid := `{"session_id":"` + claimSessionID + `","claim_request_id":"` + claimRequestID + `"}`
response := serveClaimJSON(router, "/api/v1/tasks/claim-next", valid, "application/json; charset=utf-8")
if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), strings.Repeat("a", 64)) || service.claimCommand.ClaimRequestID != claimRequestID {
t.Fatalf("claim success = %d %q command %#v", response.Code, response.Body.String(), service.claimCommand)
}
service.claimFound = false
response = serveClaimJSON(router, "/api/v1/tasks/claim-next", valid, "application/json")
if response.Code != http.StatusNoContent || response.Body.Len() != 0 {
t.Fatalf("claim empty = %d %q", response.Code, response.Body.String())
}
tests := []struct {
name, body, contentType, code string
status int
}{
{"unsupported type", valid, "text/plain", "unsupported_media_type", http.StatusUnsupportedMediaType},
{"unknown field", strings.TrimSuffix(valid, "}") + `,"device_id":"` + claimDeviceID + `"}`, "application/json", "invalid_request", http.StatusBadRequest},
{"duplicate session", `{"session_id":"` + claimSessionID + `","session_id":"` + claimSessionID + `","claim_request_id":"` + claimRequestID + `"}`, "application/json", "invalid_request", http.StatusBadRequest},
{"duplicate request", `{"session_id":"` + claimSessionID + `","claim_request_id":"` + claimRequestID + `","claim_request_id":"` + claimRequestID + `"}`, "application/json", "invalid_request", http.StatusBadRequest},
{"extra json", valid + `{}`, "application/json", "invalid_request", http.StatusBadRequest},
{"too large", strings.Repeat(" ", 4097), "application/json", "request_too_large", http.StatusRequestEntityTooLarge},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
response := serveClaimJSON(router, "/api/v1/tasks/claim-next", test.body, test.contentType)
if response.Code != test.status || response.Body.String() != `{"error":"`+test.code+`"}` {
t.Fatalf("response = %d %q", response.Code, response.Body.String())
}
})
}
invalidUTF8 := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/claim-next", bytes.NewReader([]byte{'{', 0xff, '}'}))
invalidUTF8.Header.Set("Content-Type", "application/json")
invalidResponse := httptest.NewRecorder()
router.ServeHTTP(invalidResponse, invalidUTF8)
if invalidResponse.Code != http.StatusBadRequest || invalidResponse.Body.String() != `{"error":"invalid_request"}` {
t.Fatalf("invalid UTF-8 = %d %q", invalidResponse.Code, invalidResponse.Body.String())
}
}
func TestRenewStrictBindingResponseAndFixedErrors(t *testing.T) {
authenticator := &fakeDeviceAuthenticator{principal: deviceauth.Principal{ID: claimDeviceID}}
service := &fakeTaskClaimService{renewResponse: taskclaim.RenewResponse{TaskID: claimTaskID, AttemptID: claimAttemptID, ClaimGeneration: 1, LeaseExpiresAt: "2026-08-04T01:04:00Z"}}
router, _ := newRouterWithClaimService(t, &memoryStore{}, emptyDetailStore{}, emptyEvidenceStore{}, authenticator, service)
body := `{"renew_request_id":"` + claimRenewID + `","session_id":"` + claimSessionID + `","attempt_id":"` + claimAttemptID + `","claim_generation":1,"claim_token":"` + strings.Repeat("a", 64) + `","expected_lease_expires_at":"2026-08-04T01:03:00Z"}`
response := serveClaimJSON(router, "/api/v1/tasks/"+claimTaskID+"/lease/renew", body, "application/json")
if response.Code != http.StatusOK || strings.Contains(response.Body.String(), "claim_token") || service.renewCommand.TaskID != claimTaskID {
t.Fatalf("renew response = %d %q command %#v", response.Code, response.Body.String(), service.renewCommand)
}
duplicateToken := strings.Replace(body, `"expected_lease_expires_at"`, `"claim_token":"`+strings.Repeat("a", 64)+`","expected_lease_expires_at"`, 1)
response = serveClaimJSON(router, "/api/v1/tasks/"+claimTaskID+"/lease/renew", duplicateToken, "application/json")
if response.Code != http.StatusBadRequest {
t.Fatalf("duplicate token status = %d", response.Code)
}
errorsToCodes := []struct {
err error
status int
body string
}{
{taskclaim.ErrIdempotencyConflict, http.StatusConflict, `{"error":"idempotency_conflict"}`},
{taskclaim.ErrRequiresManual, http.StatusConflict, `{"error":"claim_requires_manual"}`},
{taskclaim.ErrNotCurrent, http.StatusConflict, `{"error":"claim_not_current"}`},
{taskclaim.ErrDeviceInactive, http.StatusUnauthorized, ""},
{errors.New("database path and SQL must stay private"), http.StatusServiceUnavailable, ""},
}
for _, test := range errorsToCodes {
service.renewErr = test.err
response = serveClaimJSON(router, "/api/v1/tasks/"+claimTaskID+"/lease/renew", body, "application/json")
if response.Code != test.status || response.Body.String() != test.body {
t.Fatalf("error %v = %d %q", test.err, response.Code, response.Body.String())
}
}
}
type fakeTaskClaimService struct {
claimResponse taskclaim.ClaimResponse
claimFound bool
claimErr error
renewResponse taskclaim.RenewResponse
renewErr error
claimCommand taskclaim.ClaimCommand
renewCommand taskclaim.RenewCommand
calls int
}
func (service *fakeTaskClaimService) ClaimNext(_ context.Context, _ string, command taskclaim.ClaimCommand) (taskclaim.ClaimResponse, bool, error) {
service.calls++
service.claimCommand = command
return service.claimResponse, service.claimFound, service.claimErr
}
func (service *fakeTaskClaimService) Renew(_ context.Context, _ string, command taskclaim.RenewCommand) (taskclaim.RenewResponse, error) {
service.calls++
service.renewCommand = command
return service.renewResponse, service.renewErr
}
func serveClaimJSON(router http.Handler, path, body, contentType string) *httptest.ResponseRecorder {
request := httptest.NewRequest(http.MethodPost, path, io.NopCloser(strings.NewReader(body)))
request.Header.Set("Content-Type", contentType)
response := httptest.NewRecorder()
router.ServeHTTP(response, request)
return response
}