248 lines
10 KiB
Go
248 lines
10 KiB
Go
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
|
|
}
|