@@ -2,6 +2,7 @@ package handlers
22
33import (
44 "bytes"
5+ "container/list"
56 "encoding/binary"
67 "encoding/hex"
78 "encoding/json"
@@ -19,14 +20,15 @@ import (
1920)
2021
2122const (
22- confidentialEnvelopeSchema = 1
23- confidentialEnvelopeLimit = maxRequestBodySize + 64 * 1024
24- confidentialMaxAge = 2 * time .Minute
25- confidentialFutureSkew = 30 * time .Second
26- confidentialReplayTTL = 3 * time .Minute
27- confidentialReplayCapacity = 100_000
28- confidentialMaxEHBPFrame = 1 * 1024 * 1024 + 16
29- confidentialMaxEHBPBody = 16 * 1024 * 1024
23+ confidentialEnvelopeSchema = 1
24+ confidentialEnvelopeLimit = maxRequestBodySize + 64 * 1024
25+ confidentialMaxAge = 2 * time .Minute
26+ confidentialFutureSkew = 30 * time .Second
27+ confidentialReplayTTL = 3 * time .Minute
28+ confidentialReplayCapacity = 100_000
29+ confidentialReplayAccountCapacity = 10_000
30+ confidentialMaxEHBPFrame = 1 * 1024 * 1024 + 16
31+ confidentialMaxEHBPBody = 16 * 1024 * 1024
3032)
3133
3234// ConfidentialChatCompletionsHandler is the strict EHBP-only wrapper around
@@ -36,6 +38,7 @@ const (
3638type ConfidentialChatCompletionsHandler struct {
3739 encrypted http.Handler
3840 next http.Handler
41+ auth ports.AuthService
3942 logger ports.Logger
4043 replays * confidentialReplayCache
4144 now func () time.Time
@@ -53,18 +56,23 @@ type confidentialEnvelope struct {
5356// the enclave; a public-only identity is intentionally rejected.
5457func NewConfidentialChatCompletionsHandler (
5558 serverIdentity * identity.Identity ,
59+ auth ports.AuthService ,
5660 next http.Handler ,
5761 logger ports.Logger ,
5862) (* ConfidentialChatCompletionsHandler , error ) {
5963 if serverIdentity == nil || serverIdentity .PrivateKey () == nil {
6064 return nil , errors .New ("confidential handler requires a private EHBP identity" )
6165 }
66+ if auth == nil {
67+ return nil , errors .New ("confidential handler requires authentication" )
68+ }
6269 if next == nil {
6370 return nil , errors .New ("confidential handler requires a downstream handler" )
6471 }
6572
6673 h := & ConfidentialChatCompletionsHandler {
6774 next : next ,
75+ auth : auth ,
6876 logger : logger ,
6977 replays : newConfidentialReplayCache (confidentialReplayTTL , confidentialReplayCapacity ),
7078 now : time .Now ,
@@ -116,6 +124,21 @@ func isCanonicalEHBPKey(encoded string) bool {
116124// handleDecrypted executes only after the EHBP middleware authenticates the
117125// first encrypted frame and installs response encryption.
118126func (h * ConfidentialChatCompletionsHandler ) handleDecrypted (w http.ResponseWriter , r * http.Request ) {
127+ token , err := ExtractBearerToken (r )
128+ if err != nil {
129+ rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , err )
130+ return
131+ }
132+ auth , err := h .auth .AuthenticateAPIKey (r .Context (), token )
133+ if err != nil {
134+ rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , err )
135+ return
136+ }
137+ if auth .Account .ID == "" {
138+ rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , domain .ErrInternal ("authentication returned no account" ))
139+ return
140+ }
141+
119142 body , err := io .ReadAll (io .LimitReader (r .Body , confidentialEnvelopeLimit + 1 ))
120143 if err != nil {
121144 rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , domain .ErrInvalidField ("failed to read confidential request" ))
@@ -132,7 +155,7 @@ func (h *ConfidentialChatCompletionsHandler) handleDecrypted(w http.ResponseWrit
132155 return
133156 }
134157
135- now := h .now (). UTC ()
158+ now := h .now ()
136159 issuedAt , err := time .Parse (time .RFC3339 , envelope .IssuedAt )
137160 if err != nil {
138161 rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , domain .ErrInvalidField ("issued_at must be an RFC3339 timestamp" ))
@@ -153,18 +176,24 @@ func (h *ConfidentialChatCompletionsHandler) handleDecrypted(w http.ResponseWrit
153176 return
154177 }
155178
156- switch h .replays .checkAndStore (envelope .RequestID , now ) {
179+ switch h .replays .checkAndStore (auth . Account . ID , envelope .RequestID , h . now ) {
157180 case replayDuplicate :
158181 rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , domain .ErrConflict ("confidential request was already processed" ))
159182 return
183+ case replayAccountCapacityReached :
184+ rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , & domain.GatewayError {
185+ HTTPStatus : http .StatusTooManyRequests , Type : domain .ErrTypeRateLimit ,
186+ Code : domain .ErrCodeConfidentialLimit , Message : "Account confidential request limit reached; retry after replay entries expire" ,
187+ })
188+ return
160189 case replayCapacityReached :
161190 rejectRequest (w , r , h .logger , domain .EndpointChatCompletions , domain .ErrProviderUnavailable ("confidential replay protection" ).WithMeta (
162191 "fault" , domain .FaultNexus , "dependency" , "replay_store" , "cause" , domain .CauseReplayStoreFull ))
163192 return
164193 }
165194
166- // The existing handler remains the single implementation of bearer auth,
167- // OpenAI request validation, generation, and streaming. Only replace the
195+ // The downstream handler rechecks authentication and remains responsible for
196+ // OpenAI validation, generation, and streaming. Only replace the
168197 // decrypted envelope with its exact nested OpenAI request object.
169198 r .Body = io .NopCloser (bytes .NewReader (envelope .Request ))
170199 r .ContentLength = int64 (len (envelope .Request ))
@@ -201,46 +230,64 @@ const (
201230 replayAccepted replayResult = iota
202231 replayDuplicate
203232 replayCapacityReached
233+ replayAccountCapacityReached
204234)
205235
236+ type confidentialReplayKey struct { accountID , requestID string }
237+ type confidentialReplayEntry struct {
238+ key confidentialReplayKey
239+ expiresAt time.Time
240+ }
241+
206242type confidentialReplayCache struct {
207- mu sync.Mutex
208- ttl time.Duration
209- capacity int
210- entries map [string ]time.Time
211- nextGC time.Time
243+ // ponytail: one lock for a bounded cache; shard by account if contention is measured.
244+ mu sync.Mutex
245+ ttl time.Duration
246+ capacity int
247+ accountCapacity int
248+ entries map [confidentialReplayKey ]struct {}
249+ accounts map [string ]int
250+ expirations * list.List
212251}
213252
214253func newConfidentialReplayCache (ttl time.Duration , capacity int ) * confidentialReplayCache {
215254 return & confidentialReplayCache {
216- ttl : ttl ,
217- capacity : capacity ,
218- entries : make (map [string ]time.Time ),
255+ ttl : ttl , capacity : capacity , accountCapacity : min (capacity , confidentialReplayAccountCapacity ),
256+ entries : make (map [confidentialReplayKey ]struct {}), accounts : make (map [string ]int ), expirations : list .New (),
219257 }
220258}
221259
222- func (c * confidentialReplayCache ) checkAndStore (requestID string , now time.Time ) replayResult {
260+ func (c * confidentialReplayCache ) checkAndStore (accountID , requestID string , clock func () time.Time ) replayResult {
223261 c .mu .Lock ()
224262 defer c .mu .Unlock ()
225-
226- if expiresAt , ok := c .entries [requestID ]; ok && now .Before (expiresAt ) {
263+ now := clock () // Sample under the lock so expiry order follows insertion order.
264+ // Fixed TTLs preserve insertion order. Remove only expired entries, never
265+ // live replay protection; full-cache rejection does not scan the map.
266+ for front := c .expirations .Front (); front != nil ; front = c .expirations .Front () {
267+ entry := front .Value .(confidentialReplayEntry )
268+ if now .Before (entry .expiresAt ) {
269+ break
270+ }
271+ delete (c .entries , entry .key )
272+ c .accounts [entry .key .accountID ]--
273+ if c .accounts [entry .key .accountID ] == 0 {
274+ delete (c .accounts , entry .key .accountID )
275+ }
276+ c .expirations .Remove (front )
277+ }
278+ key := confidentialReplayKey {accountID : accountID , requestID : requestID }
279+ if _ , ok := c .entries [key ]; ok {
227280 return replayDuplicate
228281 }
229-
230- if c .nextGC .IsZero () || ! now .Before (c .nextGC ) || len (c .entries ) >= c .capacity {
231- // Freshness validation happens before this call, so an entry can safely
232- // expire after the maximum request age.
233- for id , expiresAt := range c .entries {
234- if ! now .Before (expiresAt ) {
235- delete (c .entries , id )
236- }
237- }
238- c .nextGC = now .Add (time .Minute )
282+ if c .accounts [accountID ] >= c .accountCapacity {
283+ return replayAccountCapacityReached
239284 }
240285 if len (c .entries ) >= c .capacity {
241286 return replayCapacityReached
242287 }
243- c .entries [requestID ] = now .Add (c .ttl )
288+ c .entries [key ] = struct {}{}
289+ c .accounts [accountID ]++
290+ c .expirations .PushBack (confidentialReplayEntry {key : key , expiresAt : now .Add (c .ttl )})
244291 return replayAccepted
245292}
246293
0 commit comments