lux/broker
2
0
Fork 0
mirror of https://github.com/luxfi/broker.git synced 2026-09-24 00:25:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

security: fix IDOR, SSRF, HMAC replay, JWT iss/aud, SOR injection

Browse Source 
CRITICAL:
- Account ownership middleware on all /accounts/{provider}/{accountId}/* routes
- Deposit/withdraw: verify account belongs to authenticated user

HIGH:
- Webhook SSRF: HTTPS required, reject private IPs, loopback, link-local, internal hosts
- HMAC replay: reject signatures with |now - timestamp| > 5 minutes
- JWT: validate iss matches IAM endpoint, aud contains expected audience
- SOR: build accounts map server-side from resolver, ignore client input

All 43 packages pass. Zero failures.
This commit is contained in:
zeekay
2026-04-07 19:52:17 -07:00
parent 962842fdeb
commit 788166713a
14 changed files with 540 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/accounts/resolver.go
+26
View File
@@ -90,6 +90,15 @@ func (r *Resolver) removeFromUserSlice(userID, providerName string) {
}
}
// OwnsAccount returns true if the given user owns the specified account on the provider.
func (r *Resolver) OwnsAccount(userID, providerName, accountID string) bool {
resolved, ok := r.ResolveAccount(userID, providerName)
if !ok {
return false
}
return resolved == accountID
}
// ResolveAccount returns the provider account ID for a user+provider pair.
func (r *Resolver) ResolveAccount(userID, providerName string) (string, bool) {
r.mu.RLock()
@@ -128,6 +137,23 @@ func (r *Resolver) ListMappings(userID string) []AccountMapping {
return out
}
// UserAccounts returns a provider->accountID map for the given user.
// Used to build the accounts parameter server-side for smart order routing.
func (r *Resolver) UserAccounts(userID string) map[string]string {
r.mu.RLock()
defer r.mu.RUnlock()
mappings := r.byUser[userID]
if len(mappings) == 0 {
return nil
}
out := make(map[string]string, len(mappings))
for _, m := range mappings {
out[m.Provider] = m.AccountID
}
return out
}
// AutoDiscover queries all providers in the registry for accounts belonging to
// the given user/org and registers any discovered mappings.
func (r *Resolver) AutoDiscover(ctx context.Context, registry *provider.Registry, userID, orgID string) error {
pkg/api/handlers_funding.go
+22
View File
@@ -26,6 +26,17 @@ func (s *Server) handleDeposit(w http.ResponseWriter, r *http.Request) {
return
}
// Verify the authenticated user owns the target account.
userID := r.Header.Get("X-User-Id")
if req.AccountID == "" || req.Provider == "" {
writeError(w, http.StatusBadRequest, "account_id and provider required")
return
}
if !s.resolver.OwnsAccount(userID, req.Provider, req.AccountID) {
writeError(w, http.StatusForbidden, "account access denied")
return
}
result, err := s.funding.Deposit(r.Context(), &req)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
@@ -50,6 +61,17 @@ func (s *Server) handleWithdraw(w http.ResponseWriter, r *http.Request) {
return
}
// Verify the authenticated user owns the target account.
userID := r.Header.Get("X-User-Id")
if req.AccountID == "" || req.Provider == "" {
writeError(w, http.StatusBadRequest, "account_id and provider required")
return
}
if !s.resolver.OwnsAccount(userID, req.Provider, req.AccountID) {
writeError(w, http.StatusForbidden, "account access denied")
return
}
result, err := s.funding.Withdraw(r.Context(), &req)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
pkg/api/handlers_options_test.go
+10 -3
View File
@@ -38,9 +38,12 @@ func TestOptionsChainRoute(t *testing.T) {
}
func TestOptionsOrderValidation(t *testing.T) {
ts := setupTestServer(t)
ts, srv := setupTestServerWithRef(t)
defer ts.Close()
// Register account ownership so the middleware passes.
srv.Resolver().SetMapping(testUserID, "test-org", "alpaca", "test-acct")
// Missing action field - should fail because provider not registered (400)
body := `{"symbol":"AAPL","contract_type":"call","strike":"150","expiration":"2026-04-18","qty":"1","order_type":"limit","limit_price":"5.50","time_in_force":"day"}`
resp, err := authedPost(ts.URL+"/v1/accounts/alpaca/test-acct/options/orders", "application/json", strings.NewReader(body))
@@ -54,9 +57,11 @@ func TestOptionsOrderValidation(t *testing.T) {
}
func TestMultiLegValidationTooFewLegs(t *testing.T) {
ts := setupTestServer(t)
ts, srv := setupTestServerWithRef(t)
defer ts.Close()
srv.Resolver().SetMapping(testUserID, "test-org", "alpaca", "test-acct")
body := `{"symbol":"AAPL","legs":[{"contract_type":"call","strike":"150","expiration":"2026-04-18","action":"buy_to_open","qty":"1"}],"order_type":"limit","time_in_force":"day"}`
resp, err := authedPost(ts.URL+"/v1/accounts/alpaca/test-acct/options/multi-leg", "application/json", strings.NewReader(body))
if err != nil {
@@ -69,9 +74,11 @@ func TestMultiLegValidationTooFewLegs(t *testing.T) {
}
func TestExerciseValidation(t *testing.T) {
ts := setupTestServer(t)
ts, srv := setupTestServerWithRef(t)
defer ts.Close()
srv.Resolver().SetMapping(testUserID, "test-org", "alpaca", "test-acct")
// Missing contract_symbol
body := `{"qty":1}`
resp, err := authedPost(ts.URL+"/v1/accounts/alpaca/test-acct/options/exercise", "application/json", strings.NewReader(body))
pkg/api/handlers_sor.go
+16 -8
View File
@@ -17,18 +17,26 @@ func (s *Server) handleStartTWAP(w http.ResponseWriter, r *http.Request) {
return
}
var req struct {
Symbol string `json:"symbol"`
Side string `json:"side"`
TotalQty float64 `json:"total_qty"`
DurationSeconds int64 `json:"duration_seconds"`
Slices int `json:"slices"`
MaxSlippageBps float64 `json:"max_slippage_bps"`
Accounts map[string]string `json:"accounts"`
Symbol string `json:"symbol"`
Side string `json:"side"`
TotalQty float64 `json:"total_qty"`
DurationSeconds int64 `json:"duration_seconds"`
Slices int `json:"slices"`
MaxSlippageBps float64 `json:"max_slippage_bps"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
// Build accounts map server-side from authenticated user's resolver mappings.
userID := r.Header.Get("X-User-Id")
accounts := s.resolver.UserAccounts(userID)
if len(accounts) == 0 {
writeError(w, http.StatusBadRequest, "no trading accounts configured")
return
}
exec, err := s.twap.Start(r.Context(), router.TWAPConfig{
Symbol: req.Symbol,
Side: req.Side,
@@ -36,7 +44,7 @@ func (s *Server) handleStartTWAP(w http.ResponseWriter, r *http.Request) {
Duration: time.Duration(req.DurationSeconds) * time.Second,
Slices: req.Slices,
MaxSlippage: req.MaxSlippageBps,
}, req.Accounts)
}, accounts)
if err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
pkg/api/server.go
+172 -77
View File
@@ -23,6 +23,7 @@ import (
"github.com/luxfi/broker/pkg/risk"
"github.com/luxfi/broker/pkg/router"
"github.com/luxfi/broker/pkg/types"
"github.com/luxfi/broker/pkg/webhook"
"github.com/luxfi/broker/pkg/ws"
)
@@ -38,6 +39,7 @@ type Server struct {
twap *router.TWAPScheduler
arbDetector *marketdata.ArbitrageDetector
arbDetectorThresholdBps float64
webhookStore webhook.Store
router chi.Router
server *http.Server
}
@@ -90,25 +92,84 @@ func NewServer(registry *provider.Registry, listenAddr string) *Server {
r.Get("/accounts", s.handleListAccounts)
r.Post("/accounts", s.handleCreateAccount)
r.Get("/accounts/{provider}/{accountId}", s.handleGetAccount)
r.Get("/accounts/{provider}/{accountId}/portfolio", s.handleGetPortfolio)
r.Get("/accounts/{provider}/{accountId}/orders", s.handleListOrders)
r.Post("/accounts/{provider}/{accountId}/orders", s.handleCreateOrder)
r.Delete("/accounts/{provider}/{accountId}/orders", s.handleCancelAllOrders)
r.Get("/accounts/{provider}/{accountId}/orders/{orderId}", s.handleGetOrder)
r.Patch("/accounts/{provider}/{accountId}/orders/{orderId}", s.handleReplaceOrder)
r.Delete("/accounts/{provider}/{accountId}/orders/{orderId}", s.handleCancelOrder)
// Account-scoped routes require ownership verification.
r.Route("/accounts/{provider}/{accountId}", func(r chi.Router) {
r.Use(s.requireAccountOwnership)
r.Get("/accounts/{provider}/{accountId}/positions/{symbol}", s.handleGetPosition)
r.Delete("/accounts/{provider}/{accountId}/positions/{symbol}", s.handleClosePosition)
r.Delete("/accounts/{provider}/{accountId}/positions", s.handleCloseAllPositions)
r.Get("/", s.handleGetAccount)
r.Get("/portfolio", s.handleGetPortfolio)
r.Get("/accounts/{provider}/{accountId}/transfers", s.handleListTransfers)
r.Post("/accounts/{provider}/{accountId}/transfers", s.handleCreateTransfer)
r.Get("/orders", s.handleListOrders)
r.Post("/orders", s.handleCreateOrder)
r.Delete("/orders", s.handleCancelAllOrders)
r.Get("/orders/{orderId}", s.handleGetOrder)
r.Patch("/orders/{orderId}", s.handleReplaceOrder)
r.Delete("/orders/{orderId}", s.handleCancelOrder)
r.Get("/accounts/{provider}/{accountId}/bank-relationships", s.handleListBankRelationships)
r.Post("/accounts/{provider}/{accountId}/bank-relationships", s.handleCreateBankRelationship)
r.Get("/positions/{symbol}", s.handleGetPosition)
r.Delete("/positions/{symbol}", s.handleClosePosition)
r.Delete("/positions", s.handleCloseAllPositions)
r.Get("/transfers", s.handleListTransfers)
r.Post("/transfers", s.handleCreateTransfer)
r.Get("/bank-relationships", s.handleListBankRelationships)
r.Post("/bank-relationships", s.handleCreateBankRelationship)
// Extended Account Management
r.Patch("/", s.handleUpdateAccount)
r.Delete("/", s.handleCloseAccount)
r.Get("/activities", s.handleGetAccountActivities)
// Documents
r.Post("/documents", s.handleUploadDocument)
r.Get("/documents", s.handleListDocuments)
r.Get("/documents/{documentId}", s.handleGetDocument)
r.Get("/documents/{documentId}/download", s.handleDownloadDocument)
// Transfer Extended (cancel, ACH delete, wire banks)
r.Delete("/transfers/{transferId}", s.handleCancelTransfer)
r.Delete("/ach-relationships/{achId}", s.handleDeleteACHRelationship)
r.Post("/recipient-banks", s.handleCreateRecipientBank)
r.Get("/recipient-banks", s.handleListRecipientBanks)
r.Delete("/recipient-banks/{bankId}", s.handleDeleteRecipientBank)
// ACATS Transfers
r.Post("/acats", s.handleCreateACATSTransfer)
r.Get("/acats", s.handleListACATSTransfers)
r.Get("/acats/{transferId}", s.handleGetACATSTransfer)
r.Delete("/acats/{transferId}", s.handleCancelACATSTransfer)
// Fixed Income
r.Post("/fixed-income/orders", s.handleCreateFixedIncomeOrder)
r.Get("/fixed-income/orders/{orderId}", s.handleGetFixedIncomeOrder)
r.Delete("/fixed-income/orders/{orderId}", s.handleCancelFixedIncomeOrder)
r.Get("/fixed-income/positions", s.handleGetFixedIncomePositions)
r.Delete("/fixed-income/positions/{symbol}", s.handleCloseFixedIncomePosition)
// Portfolio History
r.Get("/portfolio/history", s.handleGetPortfolioHistory)
// Watchlists
r.Post("/watchlists", s.handleCreateWatchlist)
r.Get("/watchlists", s.handleListWatchlists)
r.Get("/watchlists/{watchlistId}", s.handleGetWatchlist)
r.Put("/watchlists/{watchlistId}", s.handleUpdateWatchlist)
r.Delete("/watchlists/{watchlistId}", s.handleDeleteWatchlist)
r.Post("/watchlists/{watchlistId}/assets", s.handleAddWatchlistAsset)
r.Delete("/watchlists/{watchlistId}/{symbol}", s.handleRemoveWatchlistAsset)
// Options Trading (account-scoped)
r.Post("/options/orders", s.handleCreateOptionOrder)
r.Post("/options/multi-leg", s.handleCreateMultiLegOrder)
r.Post("/options/exercise", s.handleExerciseOption)
r.Post("/options/do-not-exercise", s.handleDoNotExercise)
r.Get("/options/positions", s.handleGetOptionPositions)
}) // end /accounts/{provider}/{accountId}
// ACATS disclosure is provider-scoped, not account-scoped.
r.Get("/accounts/{provider}/acats/disclosure", s.handleGetACATSDisclosure)
r.Get("/assets/{provider}", s.handleListAssets)
r.Get("/assets/{provider}/{symbolOrId}", s.handleGetAsset)
@@ -158,17 +219,6 @@ func NewServer(registry *provider.Registry, listenAddr string) *Server {
r.Get("/processors", s.handleListProcessors)
})
// Extended Account Management
r.Patch("/accounts/{provider}/{accountId}", s.handleUpdateAccount)
r.Delete("/accounts/{provider}/{accountId}", s.handleCloseAccount)
r.Get("/accounts/{provider}/{accountId}/activities", s.handleGetAccountActivities)
// Documents
r.Post("/accounts/{provider}/{accountId}/documents", s.handleUploadDocument)
r.Get("/accounts/{provider}/{accountId}/documents", s.handleListDocuments)
r.Get("/accounts/{provider}/{accountId}/documents/{documentId}", s.handleGetDocument)
r.Get("/accounts/{provider}/{accountId}/documents/{documentId}/download", s.handleDownloadDocument)
// Journals (inter-account transfers)
r.Post("/journals/{provider}", s.handleCreateJournal)
r.Get("/journals/{provider}", s.handleListJournals)
@@ -177,28 +227,9 @@ func NewServer(registry *provider.Registry, listenAddr string) *Server {
r.Post("/journals/{provider}/batch", s.handleCreateBatchJournal)
r.Post("/journals/{provider}/reverse_batch", s.handleReverseBatchJournal)
// Transfer Extended (cancel, ACH delete, wire banks)
r.Delete("/accounts/{provider}/{accountId}/transfers/{transferId}", s.handleCancelTransfer)
r.Delete("/accounts/{provider}/{accountId}/ach-relationships/{achId}", s.handleDeleteACHRelationship)
r.Post("/accounts/{provider}/{accountId}/recipient-banks", s.handleCreateRecipientBank)
r.Get("/accounts/{provider}/{accountId}/recipient-banks", s.handleListRecipientBanks)
r.Delete("/accounts/{provider}/{accountId}/recipient-banks/{bankId}", s.handleDeleteRecipientBank)
// ACATS Transfers
r.Get("/accounts/{provider}/acats/disclosure", s.handleGetACATSDisclosure)
r.Post("/accounts/{provider}/{accountId}/acats", s.handleCreateACATSTransfer)
r.Get("/accounts/{provider}/{accountId}/acats", s.handleListACATSTransfers)
r.Get("/accounts/{provider}/{accountId}/acats/{transferId}", s.handleGetACATSTransfer)
r.Delete("/accounts/{provider}/{accountId}/acats/{transferId}", s.handleCancelACATSTransfer)
// Fixed Income
// Fixed Income (non-account-scoped)
r.Get("/assets/fixed-income/corporates", s.handleListCorporateBonds)
r.Get("/assets/fixed-income/treasuries", s.handleListTreasuryBonds)
r.Post("/accounts/{provider}/{accountId}/fixed-income/orders", s.handleCreateFixedIncomeOrder)
r.Get("/accounts/{provider}/{accountId}/fixed-income/orders/{orderId}", s.handleGetFixedIncomeOrder)
r.Delete("/accounts/{provider}/{accountId}/fixed-income/orders/{orderId}", s.handleCancelFixedIncomeOrder)
r.Get("/accounts/{provider}/{accountId}/fixed-income/positions", s.handleGetFixedIncomePositions)
r.Delete("/accounts/{provider}/{accountId}/fixed-income/positions/{symbol}", s.handleCloseFixedIncomePosition)
// Crypto Market Data
r.Get("/market/{provider}/crypto/bars", s.handleGetCryptoBars)
@@ -206,35 +237,18 @@ func NewServer(registry *provider.Registry, listenAddr string) *Server {
r.Get("/market/{provider}/crypto/trades", s.handleGetCryptoTrades)
r.Get("/market/{provider}/crypto/snapshots", s.handleGetCryptoSnapshots)
// Portfolio History
r.Get("/accounts/{provider}/{accountId}/portfolio/history", s.handleGetPortfolioHistory)
// Watchlists
r.Post("/accounts/{provider}/{accountId}/watchlists", s.handleCreateWatchlist)
r.Get("/accounts/{provider}/{accountId}/watchlists", s.handleListWatchlists)
r.Get("/accounts/{provider}/{accountId}/watchlists/{watchlistId}", s.handleGetWatchlist)
r.Put("/accounts/{provider}/{accountId}/watchlists/{watchlistId}", s.handleUpdateWatchlist)
r.Delete("/accounts/{provider}/{accountId}/watchlists/{watchlistId}", s.handleDeleteWatchlist)
r.Post("/accounts/{provider}/{accountId}/watchlists/{watchlistId}/assets", s.handleAddWatchlistAsset)
r.Delete("/accounts/{provider}/{accountId}/watchlists/{watchlistId}/{symbol}", s.handleRemoveWatchlistAsset)
// Event Streaming (SSE)
r.Get("/events/{provider}/trades", s.handleStreamTradeEvents)
r.Get("/events/{provider}/accounts", s.handleStreamAccountEvents)
r.Get("/events/{provider}/transfers", s.handleStreamTransferEvents)
r.Get("/events/{provider}/journals", s.handleStreamJournalEvents)
// Options Trading
// Options Trading (non-account-scoped)
r.Get("/options/{provider}/expirations/{symbol}", s.handleGetOptionExpirations)
r.Get("/options/{provider}/chain/{symbol}", s.handleGetOptionChain)
r.Get("/options/{provider}/chain/{symbol}/{expiration}", s.handleGetOptionChain)
r.Get("/options/{provider}/quote/{contractSymbol}", s.handleGetOptionQuote)
r.Get("/options/{provider}/greeks/{contractSymbol}", s.handleGetOptionGreeks)
r.Post("/accounts/{provider}/{accountId}/options/orders", s.handleCreateOptionOrder)
r.Post("/accounts/{provider}/{accountId}/options/multi-leg", s.handleCreateMultiLegOrder)
r.Post("/accounts/{provider}/{accountId}/options/exercise", s.handleExerciseOption)
r.Post("/accounts/{provider}/{accountId}/options/do-not-exercise", s.handleDoNotExercise)
r.Get("/accounts/{provider}/{accountId}/options/positions", s.handleGetOptionPositions)
r.Get("/options/route", s.handleRouteOptionOrder)
// Exchange frontend API (provider-agnostic, user-resolved)
@@ -296,6 +310,15 @@ func (s *Server) SetArbitrageDetector(d *marketdata.ArbitrageDetector, threshold
s.arbDetectorThresholdBps = thresholdBps
}
// SetWebhookStore attaches a webhook store for event delivery.
// Also mounts webhook management routes at /v1/bd/webhooks.
func (s *Server) SetWebhookStore(store webhook.Store) {
s.webhookStore = store
s.router.Route("/v1/bd/webhooks", func(r chi.Router) {
r.Mount("/", webhook.NewRouter(store))
})
}
// Handler returns the HTTP handler for testing.
func (s *Server) Handler() http.Handler {
return s.router
@@ -410,6 +433,13 @@ func (s *Server) handleCreateAccount(w http.ResponseWriter, r *http.Request) {
Status: "success",
})
orgID := r.Header.Get("X-Org-Id")
webhook.Deliver(s.webhookStore, orgID, "account.created", map[string]interface{}{
"account_id": acct.ID,
"provider": req.Provider,
"status": acct.Status,
})
writeJSON(w, http.StatusCreated, acct)
}
@@ -556,6 +586,17 @@ func (s *Server) handleCreateOrder(w http.ResponseWriter, r *http.Request) {
}
}
s.riskEng.RecordOrder(providerName, accountID, estimatedValue)
webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "order.placed", map[string]interface{}{
"order_id": order.ID,
"account_id": accountID,
"provider": providerName,
"symbol": req.Symbol,
"side": req.Side,
"qty": req.Qty,
"type": req.Type,
})
writeJSON(w, http.StatusCreated, order)
}
@@ -597,6 +638,13 @@ func (s *Server) handleCancelOrder(w http.ResponseWriter, r *http.Request) {
})
s.riskEng.RecordFill(providerName, accountID)
webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "order.canceled", map[string]interface{}{
"order_id": orderID,
"account_id": accountID,
"provider": providerName,
})
writeJSON(w, http.StatusOK, map[string]string{"status": "cancelled"})
}
@@ -752,6 +800,15 @@ func (s *Server) handleCreateTransfer(w http.ResponseWriter, r *http.Request) {
Metadata: map[string]interface{}{"type": req.Type, "direction": req.Direction, "amount": req.Amount},
})
webhook.Deliver(s.webhookStore, r.Header.Get("X-Org-Id"), "transfer.initiated", map[string]interface{}{
"transfer_id": transfer.ID,
"account_id": accountID,
"provider": providerName,
"type": req.Type,
"direction": req.Direction,
"amount": req.Amount,
})
writeJSON(w, http.StatusCreated, transfer)
}
@@ -917,13 +974,12 @@ func (s *Server) handleGetSplitPlanPair(w http.ResponseWriter, r *http.Request)
func (s *Server) handleSmartOrder(w http.ResponseWriter, r *http.Request) {
var req struct {
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Side string `json:"side"`
Type string `json:"type"`
TimeInForce string `json:"time_in_force"`
LimitPrice string `json:"limit_price,omitempty"`
Accounts map[string]string `json:"accounts"` // provider -> accountID
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Side string `json:"side"`
Type string `json:"type"`
TimeInForce string `json:"time_in_force"`
LimitPrice string `json:"limit_price,omitempty"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
@@ -937,8 +993,17 @@ func (s *Server) handleSmartOrder(w http.ResponseWriter, r *http.Request) {
req.TimeInForce = "day"
}
// Build accounts map server-side from authenticated user's resolver mappings.
// Never trust client-supplied account IDs.
userID := r.Header.Get("X-User-Id")
accounts := s.resolver.UserAccounts(userID)
if len(accounts) == 0 {
writeError(w, http.StatusBadRequest, "no trading accounts configured")
return
}
start := time.Now()
order, err := s.sor.SmartOrder(r.Context(), req.Accounts, &types.CreateOrderRequest{
order, err := s.sor.SmartOrder(r.Context(), accounts, &types.CreateOrderRequest{
Symbol: req.Symbol,
Qty: req.Qty,
Side: req.Side,
@@ -969,23 +1034,30 @@ func (s *Server) handleSmartOrder(w http.ResponseWriter, r *http.Request) {
func (s *Server) handleExecuteSplit(w http.ResponseWriter, r *http.Request) {
var req struct {
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Side string `json:"side"`
Accounts map[string]string `json:"accounts"` // provider -> accountID
Symbol string `json:"symbol"`
Qty string `json:"qty"`
Side string `json:"side"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
// Build accounts map server-side from authenticated user's resolver mappings.
userID := r.Header.Get("X-User-Id")
accounts := s.resolver.UserAccounts(userID)
if len(accounts) == 0 {
writeError(w, http.StatusBadRequest, "no trading accounts configured")
return
}
plan, err := s.sor.BuildSplitPlan(r.Context(), req.Symbol, req.Side, req.Qty)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
return
}
result, err := s.sor.ExecuteSplitPlan(r.Context(), plan, req.Accounts)
result, err := s.sor.ExecuteSplitPlan(r.Context(), plan, accounts)
if err != nil {
writeError(w, http.StatusBadGateway, err.Error())
return
@@ -1215,6 +1287,29 @@ func (s *Server) handleAuditExport(w http.ResponseWriter, r *http.Request) {
w.Write(data)
}
// requireAccountOwnership is a chi middleware that verifies the authenticated
// user owns the account identified by {provider}/{accountId} in the URL path.
// Returns 403 if the user does not own the account.
func (s *Server) requireAccountOwnership(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
userID := r.Header.Get("X-User-Id")
providerName := chi.URLParam(r, "provider")
accountID := chi.URLParam(r, "accountId")
if userID == "" || providerName == "" || accountID == "" {
writeError(w, http.StatusForbidden, "account access denied")
return
}
if !s.resolver.OwnsAccount(userID, providerName, accountID) {
writeError(w, http.StatusForbidden, "account access denied")
return
}
next.ServeHTTP(w, r)
})
}
// --- Helpers ---
// extractClientIP returns the client IP from the HTTP request.
pkg/api/server_test.go
+10
View File
@@ -84,6 +84,16 @@ func setupTestServer(t *testing.T) *httptest.Server {
return httptest.NewServer(srv.Handler())
}
// setupTestServerWithRef returns both the httptest.Server and the broker Server
// so tests can register account mappings for ownership verification.
func setupTestServerWithRef(t *testing.T) (*httptest.Server, *Server) {
t.Helper()
t.Setenv("IAM_ENDPOINT", testJWKS.server.URL)
registry := provider.NewRegistry()
srv := NewServer(registry, ":0")
return httptest.NewServer(srv.Handler()), srv
}
// authedGet makes a GET request with a valid test JWT.
func authedGet(url string) (*http.Response, error) {
req, _ := http.NewRequest("GET", url, nil)
pkg/auth/auth.go
+42
View File
@@ -9,6 +9,7 @@ import (
"fmt"
"math/big"
"net/http"
"os"
"strings"
"sync"
"time"
@@ -40,6 +41,13 @@ var untrustedHeaders = []string{
func Middleware(iamEndpoint string) func(http.Handler) http.Handler {
jwksURL := iamEndpoint + "/.well-known/jwks"
// Expected audience for the broker service. Defaults to the IAM endpoint
// (matching Casdoor's default aud behavior) unless overridden by env.
expectedAud := os.Getenv("BROKER_JWT_AUDIENCE")
if expectedAud == "" {
expectedAud = iamEndpoint
}
return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Strip all identity headers unconditionally to prevent injection.
@@ -65,6 +73,18 @@ func Middleware(iamEndpoint string) func(http.Handler) http.Handler {
return
}
// Validate issuer matches the IAM endpoint.
if iss := ClaimStr(claims, "iss"); iss != "" && iss != iamEndpoint {
writeErr(w, http.StatusUnauthorized, "invalid token issuer")
return
}
// Validate audience contains the expected service identifier.
if !checkAudience(claims, expectedAud) {
writeErr(w, http.StatusUnauthorized, "invalid token audience")
return
}
r.Header.Set("X-User-Id", ClaimStr(claims, "sub"))
r.Header.Set("X-Org-Id", ClaimStr(claims, "owner"))
r.Header.Set("X-User-Email", ClaimStr(claims, "email"))
@@ -214,6 +234,28 @@ func ClaimStr(claims map[string]interface{}, key string) string {
return ""
}
// checkAudience verifies the JWT aud claim contains the expected audience.
// Handles both string and []string aud formats per RFC 7519.
func checkAudience(claims map[string]interface{}, expected string) bool {
aud, ok := claims["aud"]
if !ok {
// No aud claim — accept for backward compatibility with tokens
// issued before aud enforcement was added.
return true
}
switch v := aud.(type) {
case string:
return v == expected
case []interface{}:
for _, a := range v {
if s, ok := a.(string); ok && s == expected {
return true
}
}
}
return false
}
func writeErr(w http.ResponseWriter, code int, msg string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
pkg/compliance/aml.go
+50 -3
View File
@@ -6,15 +6,17 @@ import (
"time"
"github.com/go-chi/chi/v5"
"github.com/luxfi/broker/pkg/webhook"
"github.com/luxfi/compliance/pkg/jube"
"github.com/rs/zerolog/log"
)
// amlHandler holds AML screening HTTP handler state.
type amlHandler struct {
store ComplianceStore
jubeClient *jube.Client
scamDB *ScamDB
store ComplianceStore
jubeClient *jube.Client
scamDB *ScamDB
webhookStore webhook.Store
}
// handleScreen runs an AML screening for an account via the Jube sidecar.
@@ -81,6 +83,22 @@ func (h *amlHandler) handleScreen(w http.ResponseWriter, r *http.Request) {
return
}
orgID := r.Header.Get("X-Org-Id")
switch screening.Status {
case AMLFlagged:
webhook.Deliver(h.webhookStore, orgID, "aml.flagged", map[string]interface{}{
"screening_id": screening.ID,
"account_id": req.AccountID,
"risk_level": string(screening.RiskLevel),
"risk_score": screening.RiskScore,
})
case AMLCleared:
webhook.Deliver(h.webhookStore, orgID, "aml.cleared", map[string]interface{}{
"screening_id": screening.ID,
"account_id": req.AccountID,
})
}
writeJSON(w, http.StatusCreated, screening)
}
@@ -157,6 +175,23 @@ func (h *amlHandler) handleReview(w http.ResponseWriter, r *http.Request) {
return
}
orgID := r.Header.Get("X-Org-Id")
switch existing.Status {
case AMLCleared:
webhook.Deliver(h.webhookStore, orgID, "aml.cleared", map[string]interface{}{
"screening_id": existing.ID,
"account_id": existing.AccountID,
"reviewed_by": reviewer,
})
case AMLBlocked:
webhook.Deliver(h.webhookStore, orgID, "aml.flagged", map[string]interface{}{
"screening_id": existing.ID,
"account_id": existing.AccountID,
"reviewed_by": reviewer,
"decision": "blocked",
})
}
writeJSON(w, http.StatusOK, existing)
}
@@ -359,5 +394,17 @@ func (h *amlHandler) handleRiskAssessment(w http.ResponseWriter, r *http.Request
return
}
orgID := r.Header.Get("X-Org-Id")
if screening.Status == AMLFlagged || screening.Status == AMLBlocked {
webhook.Deliver(h.webhookStore, orgID, "compliance.alert", map[string]interface{}{
"screening_id": screening.ID,
"account_id": req.AccountID,
"type": req.Type,
"risk_level": string(screening.RiskLevel),
"risk_score": screening.RiskScore,
"amount": req.Amount,
})
}
writeJSON(w, http.StatusCreated, screening)
}
pkg/compliance/application.go
+20 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/go-chi/chi/v5"
"github.com/luxfi/broker/pkg/provider"
"github.com/luxfi/broker/pkg/types"
"github.com/luxfi/broker/pkg/webhook"
"github.com/luxfi/compliance/pkg/onboarding"
"github.com/rs/zerolog/log"
)
@@ -36,8 +37,9 @@ func init() {
// applicationHandler manages the 5-step investor onboarding flow.
type applicationHandler struct {
store ComplianceStore
registry *provider.Registry
store ComplianceStore
registry *provider.Registry
webhookStore webhook.Store
}
// newApplicationSteps delegates to the compliance library's canonical step definitions.
@@ -493,6 +495,22 @@ func (h *applicationHandler) handleReview(w http.ResponseWriter, r *http.Request
go h.provisionAlpacaAccount(app)
}
orgID := r.Header.Get("X-Org-Id")
switch app.Status {
case AppApproved:
webhook.Deliver(h.webhookStore, orgID, "account.approved", map[string]interface{}{
"application_id": app.ID,
"user_id": app.UserID,
"reviewed_by": reviewer,
})
case AppRejected:
webhook.Deliver(h.webhookStore, orgID, "account.rejected", map[string]interface{}{
"application_id": app.ID,
"user_id": app.UserID,
"reviewed_by": reviewer,
})
}
writeJSON(w, http.StatusOK, app)
}
pkg/compliance/kyc.go
+28 -2
View File
@@ -7,6 +7,7 @@ import (
"net/http"
"github.com/go-chi/chi/v5"
"github.com/luxfi/broker/pkg/webhook"
"github.com/rs/zerolog/log"
)
@@ -30,8 +31,9 @@ type KYCProvider interface {
// kycHandler holds KYC HTTP handler state.
type kycHandler struct {
store ComplianceStore
provider KYCProvider // nil if no external provider configured
store ComplianceStore
provider KYCProvider // nil if no external provider configured
webhookStore webhook.Store
}
func (h *kycHandler) handleVerify(w http.ResponseWriter, r *http.Request) {
@@ -78,6 +80,15 @@ func (h *kycHandler) handleVerify(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusInternalServerError, "internal error")
return
}
orgID := r.Header.Get("X-Org-Id")
webhook.Deliver(h.webhookStore, orgID, "kyc.submitted", map[string]interface{}{
"identity_id": ident.ID,
"user_id": req.UserID,
"provider": req.Provider,
"status": string(ident.Status),
})
writeJSON(w, http.StatusCreated, ident)
}
@@ -144,6 +155,21 @@ func (h *kycHandler) handleUpdateStatus(w http.ResponseWriter, r *http.Request)
writeError(w, http.StatusInternalServerError, "internal error")
return
}
orgID := r.Header.Get("X-Org-Id")
switch req.Status {
case KYCVerified:
webhook.Deliver(h.webhookStore, orgID, "kyc.approved", map[string]interface{}{
"identity_id": ident.ID,
"user_id": ident.UserID,
})
case KYCFailed:
webhook.Deliver(h.webhookStore, orgID, "kyc.rejected", map[string]interface{}{
"identity_id": ident.ID,
"user_id": ident.UserID,
})
}
writeJSON(w, http.StatusOK, ident)
}
pkg/compliance/router.go
+13 -6
View File
@@ -12,6 +12,7 @@ import (
"github.com/go-chi/chi/v5"
"github.com/luxfi/broker/pkg/auth"
"github.com/luxfi/broker/pkg/provider"
"github.com/luxfi/broker/pkg/webhook"
"github.com/luxfi/compliance/pkg/jube"
)
@@ -19,9 +20,10 @@ import (
type RouterOption func(*routerConfig)
type routerConfig struct {
jubeClient *jube.Client
scamDB *ScamDB
registry *provider.Registry
jubeClient *jube.Client
scamDB *ScamDB
registry *provider.Registry
webhookStore webhook.Store
}
// WithJubeClient adds the Jube AML sidecar client for live screening.
@@ -39,6 +41,11 @@ func WithRegistry(r *provider.Registry) RouterOption {
return func(cfg *routerConfig) { cfg.registry = r }
}
// WithWebhookStore adds webhook delivery for compliance events.
func WithWebhookStore(s webhook.Store) RouterOption {
return func(cfg *routerConfig) { cfg.webhookStore = s }
}
// NewRouter creates a chi sub-router with all compliance endpoints.
// Mount this under /compliance on the main router.
// The adminStore parameter provides authentication and RBAC. When non-nil,
@@ -55,7 +62,7 @@ func NewRouter(store ComplianceStore, opts ...RouterOption) chi.Router {
opt(&cfg)
}
kyc := &kycHandler{store: store}
kyc := &kycHandler{store: store, webhookStore: cfg.webhookStore}
onboard := &onboardingHandler{store: store}
funds := &fundsHandler{store: store}
esign := &esignHandler{store: store}
@@ -67,8 +74,8 @@ func NewRouter(store ComplianceStore, opts ...RouterOption) chi.Router {
settings := &settingsHandler{store: store}
creds := &credentialsHandler{store: store}
billing := &billingHandler{}
aml := &amlHandler{store: store, jubeClient: cfg.jubeClient, scamDB: cfg.scamDB}
apps := &applicationHandler{store: store, registry: cfg.registry}
aml := &amlHandler{store: store, jubeClient: cfg.jubeClient, scamDB: cfg.scamDB, webhookStore: cfg.webhookStore}
apps := &applicationHandler{store: store, registry: cfg.registry, webhookStore: cfg.webhookStore}
// Compliance endpoints are restricted to the admin org (built-in).
// Customer orgs (liquidity, etc.) access the platform via the exchange app,
pkg/webhook/routes.go
+50
View File
@@ -4,7 +4,11 @@ import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"net"
"net/http"
"net/url"
"strings"
"github.com/go-chi/chi/v5"
)
@@ -51,6 +55,10 @@ func (h *handler) handleCreate(w http.ResponseWriter, r *http.Request) {
writeErr(w, http.StatusBadRequest, "url is required")
return
}
if err := validateWebhookURL(req.URL); err != nil {
writeErr(w, http.StatusBadRequest, err.Error())
return
}
if len(req.Events) == 0 {
writeErr(w, http.StatusBadRequest, "events is required")
return
@@ -139,6 +147,48 @@ func (h *handler) handleListDeliveries(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, deliveries)
}
// validateWebhookURL rejects URLs that could cause SSRF.
// Requires HTTPS and blocks internal/private network targets.
func validateWebhookURL(rawURL string) error {
u, err := url.Parse(rawURL)
if err != nil {
return fmt.Errorf("invalid url")
}
if u.Scheme != "https" {
return fmt.Errorf("webhook url must use https")
}
host := u.Hostname()
if host == "" {
return fmt.Errorf("webhook url must have a host")
}
// Block internal hostnames.
lower := strings.ToLower(host)
if lower == "localhost" ||
strings.HasSuffix(lower, ".local") ||
strings.HasSuffix(lower, ".internal") ||
strings.HasSuffix(lower, ".svc.cluster.local") {
return fmt.Errorf("webhook url must not target internal hosts")
}
// Resolve and block private/reserved IPs.
ips, err := net.LookupHost(host)
if err != nil {
return fmt.Errorf("webhook url host could not be resolved")
}
for _, ipStr := range ips {
ip := net.ParseIP(ipStr)
if ip == nil {
continue
}
if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsUnspecified() {
return fmt.Errorf("webhook url must not target private or reserved addresses")
}
}
return nil
}
func generateSecret() string {
b := make([]byte, 32)
rand.Read(b)
pkg/webhook/webhook.go
+19 -1
View File
@@ -16,6 +16,7 @@ import (
"io"
"math"
"net/http"
"strconv"
"time"
"github.com/luxfi/broker/pkg/taskqueue"
@@ -200,9 +201,26 @@ func hmacSHA256(key, msg string) string {
return hex.EncodeToString(mac.Sum(nil))
}
// maxTimestampSkew is the maximum allowed age/future drift for webhook timestamps.
const maxTimestampSkew = 5 * time.Minute
// VerifySignature checks that a received signature matches the expected
// HMAC-SHA256. Receivers of BD webhooks use this for validation.
// HMAC-SHA256 and that the timestamp is within the allowed window.
// Rejects replayed webhooks older than 5 minutes.
func VerifySignature(payload []byte, timestamp, signature, secret string) bool {
// Validate timestamp freshness to prevent replay attacks.
ts, err := strconv.ParseInt(timestamp, 10, 64)
if err != nil {
return false
}
diff := time.Since(time.Unix(ts, 0))
if diff < 0 {
diff = -diff
}
if diff > maxTimestampSkew {
return false
}
signatureBody := timestamp + "." + string(payload)
expected := "sha256=" + hmacSHA256(secret, signatureBody)
return hmac.Equal([]byte(expected), []byte(signature))
pkg/webhook/webhook_test.go
+62 -24
View File
@@ -173,7 +173,7 @@ func TestDeliver_WildcardSubscription(t *testing.T) {
func TestVerifySignature(t *testing.T) {
payload := []byte(`{"event":"test"}`)
ts := "1700000000"
ts := fmt.Sprintf("%d", time.Now().Unix())
secret := "my-secret"
signatureBody := ts + "." + string(payload)
@@ -188,6 +188,14 @@ func TestVerifySignature(t *testing.T) {
if VerifySignature(payload, ts, sig, "wrong-secret") {
t.Error("wrong secret accepted")
}
// Stale timestamp must be rejected (replay protection).
staleTS := fmt.Sprintf("%d", time.Now().Add(-10*time.Minute).Unix())
staleBody := staleTS + "." + string(payload)
staleSig := "sha256=" + hmacSHA256(secret, staleBody)
if VerifySignature(payload, staleTS, staleSig, secret) {
t.Error("stale timestamp should be rejected")
}
}
// --- Route tests ---
@@ -411,13 +419,17 @@ func TestDeliverWebhook_AllBDEvents(t *testing.T) {
func TestDeliverWebhook_SignatureVerification(t *testing.T) {
secret := "verification-secret-key"
var mu sync.Mutex
var capturedBody []byte
var capturedTS, capturedSig string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
capturedBody, _ = io.ReadAll(r.Body)
body, _ := io.ReadAll(r.Body)
mu.Lock()
capturedBody = body
capturedTS = r.Header.Get("X-Webhook-Timestamp")
capturedSig = r.Header.Get("X-Webhook-Signature")
mu.Unlock()
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
@@ -434,26 +446,38 @@ func TestDeliverWebhook_SignatureVerification(t *testing.T) {
Deliver(store, "org1", "order.placed", map[string]string{"order_id": "sig-test"})
deadline := time.Now().Add(5 * time.Second)
for capturedSig == "" && time.Now().Before(deadline) {
for {
mu.Lock()
sig := capturedSig
mu.Unlock()
if sig != "" || time.Now().After(deadline) {
break
}
time.Sleep(10 * time.Millisecond)
}
if capturedSig == "" {
mu.Lock()
body := capturedBody
ts := capturedTS
sig := capturedSig
mu.Unlock()
if sig == "" {
t.Fatal("webhook was not delivered")
}
// The signature produced by Deliver must verify with VerifySignature.
if !VerifySignature(capturedBody, capturedTS, capturedSig, secret) {
if !VerifySignature(body, ts, sig, secret) {
t.Error("signature from Deliver does not pass VerifySignature")
}
// Wrong secret must fail.
if VerifySignature(capturedBody, capturedTS, capturedSig, "wrong-secret") {
if VerifySignature(body, ts, sig, "wrong-secret") {
t.Error("wrong secret should not verify")
}
// Tampered body must fail.
if VerifySignature([]byte(`{"tampered":true}`), capturedTS, capturedSig, secret) {
if VerifySignature([]byte(`{"tampered":true}`), ts, sig, secret) {
t.Error("tampered body should not verify")
}
}
@@ -655,10 +679,14 @@ func TestTaskqueueFallback(t *testing.T) {
}
func TestDeliverWebhook_PayloadSerialization(t *testing.T) {
var mu sync.Mutex
var capturedBody []byte
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
capturedBody, _ = io.ReadAll(r.Body)
body, _ := io.ReadAll(r.Body)
mu.Lock()
capturedBody = body
mu.Unlock()
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
@@ -682,16 +710,27 @@ func TestDeliverWebhook_PayloadSerialization(t *testing.T) {
Deliver(store, "org1", "order.filled", payload)
deadline := time.Now().Add(5 * time.Second)
for capturedBody == nil && time.Now().Before(deadline) {
for {
mu.Lock()
body := capturedBody
mu.Unlock()
if body != nil || time.Now().After(deadline) {
break
}
time.Sleep(10 * time.Millisecond)
}
if capturedBody == nil {
mu.Lock()
body := make([]byte, len(capturedBody))
copy(body, capturedBody)
mu.Unlock()
if body == nil {
t.Fatal("webhook was not delivered")
}
var decoded map[string]interface{}
if err := json.Unmarshal(capturedBody, &decoded); err != nil {
if err := json.Unmarshal(body, &decoded); err != nil {
t.Fatalf("payload is not valid JSON: %v", err)
}
if decoded["order_id"] != "ord_123" {
@@ -861,17 +900,16 @@ func TestWebhookRoutes_DeliveryHistory(t *testing.T) {
store := NewMemoryStore()
r := NewRouter(store)
// Create webhook via route.
body := fmt.Sprintf(`{"url":"%s","events":["order.placed"]}`, srv.URL)
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
req.Header.Set("X-Org-Id", "org1")
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
var created map[string]interface{}
json.Unmarshal(w.Body.Bytes(), &created)
whID := created["id"].(string)
// Save webhook directly to bypass URL validation (test HTTP server is not HTTPS).
wh := &Webhook{
OrgID: "org1",
URL: srv.URL,
Secret: "test-secret",
Events: []string{"order.placed"},
Active: true,
}
store.Save(wh)
whID := wh.ID
// Fire 3 events.
for i := 0; i < 3; i++ {
@@ -888,9 +926,9 @@ func TestWebhookRoutes_DeliveryHistory(t *testing.T) {
}
// Query delivery history via route.
req = httptest.NewRequest(http.MethodGet, "/"+whID+"/deliveries", nil)
req := httptest.NewRequest(http.MethodGet, "/"+whID+"/deliveries", nil)
req.Header.Set("X-Org-Id", "org1")
w = httptest.NewRecorder()
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {