feat(platform): forward protected requests to run
This commit is contained in:
npc0-hue
@@ -0,0 +1,247 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"browser.local/platform/domain"
|
||||
"browser.local/platform/validator"
|
||||
)
|
||||
|
||||
const protectedRequestMaxTimeoutSeconds = 120
|
||||
|
||||
type protectedRequestPayload struct {
|
||||
commandID string
|
||||
kind string
|
||||
transportKey string
|
||||
targetKey string
|
||||
requestText string
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// protectedRequestBroker keeps opaque request text out of durable jobs and
|
||||
// bridge records. It releases a payload exactly once to a current Run lease.
|
||||
type protectedRequestBroker struct {
|
||||
mu sync.Mutex
|
||||
now func() time.Time
|
||||
payloads map[string]protectedRequestPayload
|
||||
}
|
||||
|
||||
func newProtectedRequestBroker(now func() time.Time) *protectedRequestBroker {
|
||||
return &protectedRequestBroker{now: now, payloads: map[string]protectedRequestPayload{}}
|
||||
}
|
||||
|
||||
func (broker *protectedRequestBroker) Put(jobID string, payload protectedRequestPayload) error {
|
||||
broker.mu.Lock()
|
||||
defer broker.mu.Unlock()
|
||||
broker.pruneLocked()
|
||||
if _, exists := broker.payloads[jobID]; exists {
|
||||
return validationError("protected request idempotency key is already pending")
|
||||
}
|
||||
broker.payloads[jobID] = payload
|
||||
return nil
|
||||
}
|
||||
|
||||
func (broker *protectedRequestBroker) Consume(jobID string) (protectedRequestPayload, error) {
|
||||
broker.mu.Lock()
|
||||
defer broker.mu.Unlock()
|
||||
broker.pruneLocked()
|
||||
payload, exists := broker.payloads[jobID]
|
||||
if !exists {
|
||||
return protectedRequestPayload{}, validationError("protected request input is unavailable")
|
||||
}
|
||||
delete(broker.payloads, jobID)
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func (broker *protectedRequestBroker) Delete(jobID string) {
|
||||
broker.mu.Lock()
|
||||
defer broker.mu.Unlock()
|
||||
delete(broker.payloads, jobID)
|
||||
}
|
||||
|
||||
func (broker *protectedRequestBroker) pruneLocked() {
|
||||
stamp := broker.now()
|
||||
for jobID, payload := range broker.payloads {
|
||||
if !stamp.Before(payload.expiresAt) {
|
||||
delete(broker.payloads, jobID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func protectedRequestCapability(kind string) (string, string, error) {
|
||||
switch kind {
|
||||
case "sql":
|
||||
return domain.JobCapabilityRemoteRunProtectedSQL, "protected-sql", nil
|
||||
case "rcon":
|
||||
return domain.JobCapabilityRemoteRunProtectedRCON, "protected-rcon", nil
|
||||
case "program":
|
||||
return domain.JobCapabilityRemoteRunProgram, "protected-program", nil
|
||||
default:
|
||||
return "", "", validationError("protected request kind is unsupported")
|
||||
}
|
||||
}
|
||||
|
||||
func redactedProtectedRequestPayload(declaration *domain.GameClientBridgeProtectedRequestDeclaration) map[string]any {
|
||||
return map[string]any{declaration.TextField: "redacted"}
|
||||
}
|
||||
|
||||
func (svc *CoreService) dispatchProtectedRequest(command domain.GameClientBridgeCommand, declaration domain.GameClientBridgeCommandDeclaration, payload map[string]any) error {
|
||||
if declaration.ProtectedRequest == nil || declaration.TimeoutSeconds < 1 || declaration.TimeoutSeconds > protectedRequestMaxTimeoutSeconds {
|
||||
return validationError("protected request timeout is out of bounds")
|
||||
}
|
||||
requestText, _ := payload[declaration.ProtectedRequest.TextField].(string)
|
||||
capability, adapterKind, err := protectedRequestCapability(declaration.ProtectedRequest.Kind)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
jobID := command.RunJobID
|
||||
if jobID == "" {
|
||||
return validationError("protected request job binding is missing")
|
||||
}
|
||||
if err := svc.protectedRequests.Put(jobID, protectedRequestPayload{commandID: command.ID, kind: declaration.ProtectedRequest.Kind, transportKey: declaration.ProtectedRequest.TransportKey, targetKey: declaration.ProtectedRequest.TargetKey, requestText: requestText, expiresAt: command.ExpiresAt}); err != nil {
|
||||
return err
|
||||
}
|
||||
job := domain.Job{
|
||||
ID: jobID,
|
||||
ServerInstanceID: command.ServerInstanceID,
|
||||
RunEndpointID: mustProtectedRequestRunEndpoint(svc, command.ServerInstanceID),
|
||||
Capability: capability,
|
||||
TargetKey: declaration.ProtectedRequest.TargetKey,
|
||||
InputRef: "input://protected-request/" + command.ID,
|
||||
IdempotencyKey: "protected-request:" + command.ID,
|
||||
Progress: domain.JobProgress{Percent: 0, Message: "protected request queued"},
|
||||
RetryPolicy: domain.JobRetryPolicy{MaxAttempts: 1, InitialBackoffSeconds: 1, MaxBackoffSeconds: 1},
|
||||
ExecutionInput: domain.JobExecutionInput{
|
||||
WorkspaceScope: command.ProfileKey,
|
||||
RemoteAdapterKey: declaration.ProtectedRequest.TransportKey,
|
||||
RemoteAdapterKind: adapterKind,
|
||||
TimeoutSeconds: declaration.TimeoutSeconds,
|
||||
PluginID: command.PluginID,
|
||||
},
|
||||
}
|
||||
if job.RunEndpointID == "" {
|
||||
svc.protectedRequests.Delete(jobID)
|
||||
return validationError("protected request server binding is unavailable")
|
||||
}
|
||||
created, err := svc.CreateJob(job)
|
||||
if err != nil {
|
||||
svc.protectedRequests.Delete(jobID)
|
||||
return err
|
||||
}
|
||||
if created.ID != jobID || created.ServerInstanceID != job.ServerInstanceID || created.Capability != capability || created.TargetKey != job.TargetKey || created.ExecutionInput.RemoteAdapterKey != job.ExecutionInput.RemoteAdapterKey || created.ExecutionInput.RemoteAdapterKind != adapterKind {
|
||||
svc.protectedRequests.Delete(jobID)
|
||||
return validationError("protected request idempotency key is already bound")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func mustProtectedRequestRunEndpoint(svc *CoreService, serverInstanceID string) string {
|
||||
instance, err := svc.store.ServerInstances().Get(serverInstanceID)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return instance.RunEndpointID
|
||||
}
|
||||
|
||||
func (svc *CoreService) GetProtectedRequestExecutionInput(request domain.ProtectedRequestExecutionInputRequest) (domain.ProtectedRequestExecutionInput, error) {
|
||||
if err := validator.ValidateProtectedRequestExecutionInputRequest(request); err != nil {
|
||||
return domain.ProtectedRequestExecutionInput{}, err
|
||||
}
|
||||
job, err := svc.activeFencedInputJob(request.RunEndpointID, request.SessionToken, request.JobID, request.LeaseToken, request.Attempt)
|
||||
if err != nil {
|
||||
return domain.ProtectedRequestExecutionInput{}, err
|
||||
}
|
||||
if request.FencingToken != uint64(job.Attempt) || !isProtectedRequestCapability(job.Capability) || job.RetryPolicy.MaxAttempts != 1 || !strings.HasPrefix(job.InputRef, "input://protected-request/") {
|
||||
return domain.ProtectedRequestExecutionInput{}, validationError("job is not a fenced protected request")
|
||||
}
|
||||
commands, err := svc.store.GameClientBridgeCommands().List(domain.GameClientBridgeCommandFilter{ServerInstanceID: job.ServerInstanceID})
|
||||
if err != nil {
|
||||
return domain.ProtectedRequestExecutionInput{}, err
|
||||
}
|
||||
var command domain.GameClientBridgeCommand
|
||||
for _, candidate := range commands {
|
||||
if candidate.RunJobID == job.ID {
|
||||
command = candidate
|
||||
break
|
||||
}
|
||||
}
|
||||
if command.ID == "" || command.ApprovalState != domain.GameClientBridgeApprovalApproved || command.State != domain.GameClientBridgeCommandPending || !command.ExpiresAt.After(svc.now()) {
|
||||
return domain.ProtectedRequestExecutionInput{}, validationError("protected request is not currently authorized")
|
||||
}
|
||||
payload, err := svc.protectedRequests.Consume(job.ID)
|
||||
if err != nil {
|
||||
return domain.ProtectedRequestExecutionInput{}, err
|
||||
}
|
||||
capability, adapterKind, capabilityErr := protectedRequestCapability(payload.kind)
|
||||
if capabilityErr != nil || capability != job.Capability || payload.targetKey != job.TargetKey || payload.transportKey != job.ExecutionInput.RemoteAdapterKey || adapterKind != job.ExecutionInput.RemoteAdapterKind {
|
||||
return domain.ProtectedRequestExecutionInput{}, validationError("protected request logical binding is invalid")
|
||||
}
|
||||
return domain.CopyProtectedRequestExecutionInput(domain.ProtectedRequestExecutionInput{JobID: job.ID, ServerInstanceID: job.ServerInstanceID, RunEndpointID: job.RunEndpointID, FencingToken: request.FencingToken, Authorized: true, ApprovalState: "approved", QueueState: "claimed", ExpiresAt: payload.expiresAt, Kind: payload.kind, TransportKey: payload.transportKey, TargetKey: payload.targetKey, RequestText: payload.requestText}), nil
|
||||
}
|
||||
|
||||
func isProtectedRequestCapability(capability string) bool {
|
||||
_, _, err := protectedRequestCapabilityForCapability(capability)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
func protectedRequestCapabilityForCapability(capability string) (string, string, error) {
|
||||
switch capability {
|
||||
case domain.JobCapabilityRemoteRunProtectedSQL:
|
||||
return "sql", "protected-sql", nil
|
||||
case domain.JobCapabilityRemoteRunProtectedRCON:
|
||||
return "rcon", "protected-rcon", nil
|
||||
case domain.JobCapabilityRemoteRunProgram:
|
||||
return "program", "protected-program", nil
|
||||
default:
|
||||
return "", "", validationError("job is not a protected request")
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *CoreService) projectProtectedRequestJobResult(job domain.Job, result domain.RunJobResult, stamp time.Time) error {
|
||||
if !isProtectedRequestCapability(job.Capability) {
|
||||
return nil
|
||||
}
|
||||
svc.protectedRequests.Delete(job.ID)
|
||||
svc.bridgeMu.Lock()
|
||||
defer svc.bridgeMu.Unlock()
|
||||
commands, err := svc.store.GameClientBridgeCommands().List(domain.GameClientBridgeCommandFilter{ServerInstanceID: job.ServerInstanceID})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, command := range commands {
|
||||
if command.RunJobID != job.ID || isTerminalGameClientBridgeCommandState(command.State) {
|
||||
continue
|
||||
}
|
||||
switch result.State {
|
||||
case domain.JobStateSucceeded:
|
||||
command.State = domain.GameClientBridgeCommandSucceeded
|
||||
command.Result.Status = domain.GameClientBridgeResultSucceeded
|
||||
case domain.JobStateCancelled:
|
||||
command.State = domain.GameClientBridgeCommandCancelled
|
||||
command.Result.Status = domain.GameClientBridgeResultCancelled
|
||||
case domain.JobStateFailed:
|
||||
command.State = domain.GameClientBridgeCommandFailed
|
||||
command.Result.Status = domain.GameClientBridgeResultFailed
|
||||
if result.ErrorCode == "protected_request_unknown" || strings.HasSuffix(result.ExecutionResult.Kind, ".unknown") {
|
||||
command.State = domain.GameClientBridgeCommandUnknown
|
||||
command.Result.Status = domain.GameClientBridgeResultUnknown
|
||||
}
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
command.Result.Summary = "protected request completed by Run"
|
||||
command.Result.CompletedBy = "run"
|
||||
command.Result.CompletedAt = stamp
|
||||
command.CompletedAt = stamp
|
||||
command.UpdatedAt = stamp
|
||||
auditID, auditErr := svc.recordAuditEventWithID("run", "game-client-bridge.command.result", "game-client-bridge-command", command.ID, domain.AuditResultSuccess, "Run recorded protected bridge command result")
|
||||
if auditErr != nil {
|
||||
return auditErr
|
||||
}
|
||||
command.AuditReferences = append(command.AuditReferences, auditID)
|
||||
return svc.store.GameClientBridgeCommands().Update(command)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user