package requester import ( "cmp" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "os" "regexp" "slices" "strings" "sync" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market" contractmarket "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise" "github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant" "github.com/johnandersen777/socialweb-computer-kcp/internal/sshtransport" ) const ( heldContractTTLSeconds = 691200 transientContractTTLSeconds = 3600 ) var vmNamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`) var localeNames = map[string]bool{ "LC_ALL": true, "LC_CTYPE": true, "LC_NUMERIC": true, "LC_TIME": true, "LC_COLLATE": true, "LC_MONETARY": true, "LC_MESSAGES": true, "LC_PAPER": true, "LC_NAME": true, "LC_ADDRESS": true, "LC_TELEPHONE": true, "LC_MEASUREMENT": true, "LC_IDENTIFICATION": true, } type SessionTarget struct { VMName string Address string HostKeys sshtransport.GuestHostKeys } func (q *Requester) RunCommand(ctx context.Context, account sshtransport.AuthorizedAccount, command string, env map[string]string, io sshtransport.CommandIO) error { if io == nil { return errors.New("requester: the session has no command io") } if strings.TrimSpace(account.DID) != q.repoDID { return commandFailed(io, "provisioning failed: ", fmt.Errorf("this requester rents only for %s, not for %s", q.repoDID, account.DID)) } lc := lcEnv(env) atirc := sessionWantsAtirc(command, lc) delete(lc, SessionAtircEnv) delete(lc, sessionAtircTokenEnv) if name := strings.TrimSpace(lc["LC_SESSION"]); name != "" { return q.runHeldCommand(ctx, account, name, command, lc, io) } return q.runTransientCommand(ctx, account, command, lc, atirc, io) } func commandFailed(io sshtransport.CommandIO, prefix string, err error) error { io.WriteErr([]byte(prefix + err.Error() + "\n")) io.Exit(1) return err } func (q *Requester) sessionPolicy(lc map[string]string) (string, map[string]string, error) { policyName := strings.TrimSpace(lc["LC_POLICY"]) switch strings.ToLower(policyName) { case "none", "off": return "", nil, fmt.Errorf("LC_POLICY %q: a policy always runs; name one, or open", policyName) } policyName = cmp.Or(policyName, q.cfg.PolicyWorkflow, DefaultPolicyWorkflow) policyArgs := map[string]string{} if raw := strings.TrimSpace(lc["LC_POLICY_ARGS"]); raw != "" { parsed, err := participant.ParsePolicyArgs(raw) if err != nil { return "", nil, fmt.Errorf("LC_POLICY_ARGS: %w", err) } policyArgs = parsed } return policyName, policyArgs, nil } func (q *Requester) sessionVMName(lc map[string]string) (string, error) { vmName := strings.TrimSpace(lc["LC_VM_NAME"]) if vmName == "" { vmName = q.generatedVMName() } if len(vmName) > 63 || !vmNamePattern.MatchString(vmName) { return "", fmt.Errorf("LC_VM_NAME %q is not a DNS-label-shaped name", vmName) } return vmName, nil } func (q *Requester) guestDialer() sshtransport.Dialer { if socket := strings.TrimSpace(q.cfg.Spec.IrohSocket); socket != "" { return sshtransport.UnixSocketDialer{Socket: socket, Log: q.log} } return sshtransport.DumbpipeDialer{Path: "dumbpipe", Log: q.log} } func (q *Requester) runTransientCommand(ctx context.Context, account sshtransport.AuthorizedAccount, command string, lc map[string]string, atirc bool, io sshtransport.CommandIO) error { policyName, policyArgs, err := q.sessionPolicy(lc) if err != nil { return commandFailed(io, "provisioning failed: ", err) } vmName, err := q.sessionVMName(lc) if err != nil { return commandFailed(io, "provisioning failed: ", err) } sessionCtx, cancel := context.WithTimeout(ctx, q.cfg.SessionCap) defer cancel() ctx = sessionCtx contractName := transientContractName(account.DID, vmName) defer q.trackTransient(contractName, cancel)() publicKey, privateKeyPath, err := sshtransport.GenerateGuestKeypair(q.guestKeyDir(), vmName) if err != nil { return commandFailed(io, "provisioning failed: ", err) } defer sshtransport.RemoveGuestKeypair(privateKeyPath) agent := "" if sub, ok := sessionAgentRun(command); ok && atirc { if agent, err = agentSubhandle(sub, q.handle); err != nil { return commandFailed(io, "atirc agent run: ", err) } } userData := "" if atirc { if userData, err = sessionAtircUserData(defaultUserData([]string{publicKey})); err != nil { return commandFailed(io, "provisioning failed: ", fmt.Errorf("the guest atirc agent: %w", err)) } } keep := truthy(lc["LC_KEEP_VM"]) var labels map[string]string if keep { labels = map[string]string{keepVMLabel: "true"} } target, err := q.provisionHeldContract(ctx, contractName, vmName, account.DID, publicKey, policyName, policyArgs, userData, transientContractTTLSeconds, labels) if err != nil { q.releaseHeldContract(contractName) return commandFailed(io, "provisioning failed: ", err) } if !keep { defer q.releaseHeldContract(contractName) } ticket, err := ValidateTicket(target.Address) if err != nil { return commandFailed(io, "guest session failed: ", err) } provider, err := sshtransport.NewGuestSessionProvider(q.guestDialer(), q.log) if err != nil { return commandFailed(io, "guest session failed: ", err) } if !provider.PollReady(ctx, privateKeyPath, ticket, DefaultVMReadyTimeout, target.HostKeys) { return commandFailed(io, "guest session failed: ", errors.New("the guest ssh did not answer")) } minted, err := q.sessionAtirc(ctx, account, atirc, agent, q.contractState(contractName), io) if err != nil { io.Exit(1) return err } defer q.revokeSessionAtirc(ctx, minted) switch { case minted != nil && agent != "": command = atircCommand(sessionAgentCommand) case minted != nil: command = atircCommand(command) } code := provider.RunSessionEnv(ctx, privateKeyPath, ticket, renderExecCommand(command, lc), minted.env(), io, target.HostKeys) io.Exit(code) return nil } type heldSession struct { mu sync.Mutex name string vmName string privateKeyPath string contract string target *SessionTarget timer *time.Timer generation uint64 } func (h *heldSession) stopTimer() { if h.timer != nil { h.timer.Stop() h.timer = nil } h.generation++ } func (h *heldSession) clear() { h.name = "" h.target = nil h.privateKeyPath = "" h.contract = "" } func (q *Requester) heldSession(key string) *heldSession { q.heldMu.Lock() defer q.heldMu.Unlock() held := q.heldSessions[key] if held == nil { held = &heldSession{} q.heldSessions[key] = held } return held } func (q *Requester) armHeldSession(key string, held *heldSession) { q.armHeldSessionFor(key, held, q.cfg.SessionIdle) held.mu.Lock() contract, name, keyPath := held.contract, held.name, held.privateKeyPath held.mu.Unlock() if contract != "" { q.recordHeld(contract, name, keyPath, q.cfg.SessionIdle) } } func (q *Requester) armHeldSessionFor(key string, held *heldSession, after time.Duration) { held.mu.Lock() if held.timer != nil { held.timer.Stop() } held.generation++ generation := held.generation held.timer = time.AfterFunc(after, func() { q.releaseHeldSession(key, held, generation, "idle") }) held.mu.Unlock() } func (q *Requester) recordHeld(contract, name, keyPath string, after time.Duration) { ctx, cancel := context.WithTimeout(context.WithoutCancel(q.runContext()), heldRecordTimeout) defer cancel() record := &v1alpha1.ComputeContractHeldSession{Name: name, GuestKeyPath: keyPath, Until: q.now().Add(after).UTC().Format(time.RFC3339)} if err := q.deps.Intents.MergeComputeContractStatus(ctx, contract, v1alpha1.ComputeContractStatus{HeldSession: record}); err != nil { q.log.Warn("requester_held_session_expiry_unrecorded", "contract", contract, "until", record.Until, "err", err) } } func (q *Requester) keepHeldSessions(reason string) { q.heldMu.Lock() held := make([]*heldSession, 0, len(q.heldSessions)) for _, session := range q.heldSessions { held = append(held, session) } q.heldMu.Unlock() for _, session := range held { session.mu.Lock() session.stopTimer() contract := session.contract session.mu.Unlock() if contract != "" { q.log.Info("requester_held_session_kept", "contract", contract, "reason", reason) } } } func isHeldContractName(name string) bool { rest, ok := strings.CutPrefix(name, heldContractPrefix) if !ok || len(rest) != 20 || strings.HasPrefix(name, transientContractPrefix) { return false } _, err := hex.DecodeString(rest) return err == nil } func (q *Requester) sweepHeldContract(contract v1alpha1.ComputeContract) { record := contract.Status.HeldSession if record == nil { record = &v1alpha1.ComputeContractHeldSession{} } until, err := time.Parse(time.RFC3339, record.Until) name, keyPath := record.Name, record.GuestKeyPath reason := "" switch { case err != nil: reason = "the held session records no expiry" case !q.now().Before(until): reason = "the held session expired at " + until.UTC().Format(time.RFC3339) case name == "" || keyPath == "": reason = "the held session records no session name or guest key" } var target *SessionTarget if reason == "" { found, ready, terr := contractTarget(contract) switch { case terr != nil || !ready: reason = "the held session has no guest address and host keys" case !strings.HasPrefix(keyPath, q.guestKeyDir()+string(os.PathSeparator)): reason = "the held session's guest key is outside the guest key dir" default: if _, serr := os.Stat(keyPath); serr != nil { reason = "the held session's guest key is gone" } target = found } } if reason != "" { q.log.Info("requester_stale_session_release", "contract", contract.Name, "phase", contract.Status.Phase, "reason", reason) q.releaseHeldContract(contract.Name) if keyPath != "" && strings.HasPrefix(keyPath, q.guestKeyDir()+string(os.PathSeparator)) { _ = sshtransport.RemoveGuestKeypair(keyPath) } return } key := strings.TrimSpace(contract.Spec.OwnerDid) + "\x00" + name held := q.heldSession(key) held.mu.Lock() if held.target != nil { held.mu.Unlock() return } held.name, held.vmName, held.privateKeyPath, held.target, held.contract = name, target.VMName, keyPath, target, contract.Name held.mu.Unlock() q.armHeldSessionFor(key, held, until.Sub(q.now())) q.log.Info("requester_held_session_adopted", "contract", contract.Name, "until", until.UTC().Format(time.RFC3339)) } func (q *Requester) releaseHeldSession(key string, held *heldSession, generation uint64, reason string) { held.mu.Lock() if generation != 0 && generation != held.generation { held.mu.Unlock() return } if held.timer != nil { held.timer.Stop() held.timer = nil } held.generation++ q.heldMu.Lock() if q.heldSessions[key] == held { delete(q.heldSessions, key) } q.heldMu.Unlock() privateKeyPath := held.privateKeyPath contract := held.contract held.clear() held.mu.Unlock() if contract != "" { q.releaseHeldContract(contract) q.log.Info("requester_held_session_released", "contract", contract, "reason", reason) } _ = sshtransport.RemoveGuestKeypair(privateKeyPath) } func (q *Requester) runHeldCommand(ctx context.Context, account sshtransport.AuthorizedAccount, name, command string, lc map[string]string, io sshtransport.CommandIO) error { policyName, policyArgs, err := q.sessionPolicy(lc) if err != nil { return commandFailed(io, "provisioning failed: ", err) } vmName, err := q.sessionVMName(lc) if err != nil { return commandFailed(io, "provisioning failed: ", err) } sessionCtx, cancel := context.WithTimeout(ctx, q.cfg.SessionCap) defer cancel() ctx = sessionCtx provider, err := sshtransport.NewGuestSessionProvider(q.guestDialer(), q.log) if err != nil { return commandFailed(io, "guest session failed: ", err) } contractName := heldContractName(account.DID, name) provision := func() (*SessionTarget, string, error) { publicKey, privateKeyPath, err := sshtransport.GenerateGuestKeypair(q.guestKeyDir(), vmName) if err != nil { return nil, "", err } target, err := q.provisionHeldContract(ctx, contractName, vmName, account.DID, publicKey, policyName, policyArgs, "", heldContractTTLSeconds, nil) if err != nil { q.releaseHeldContract(contractName) _ = sshtransport.RemoveGuestKeypair(privateKeyPath) return nil, "", err } return target, privateKeyPath, nil } key := account.DID + "\x00" + name held := q.heldSession(key) var reused bool var target *SessionTarget var privateKeyPath, contract string for { held.mu.Lock() held.stopTimer() reused = held.target != nil target, privateKeyPath, contract = held.target, held.privateKeyPath, held.contract if !reused { break } held.mu.Unlock() policyCtx, policyCancel := context.WithTimeout(ctx, HeldSessionPolicyTimeout) err := q.heldSessionAllowed(policyCtx, contract, policyName) policyCancel() held.mu.Lock() if held.target != target || held.contract != contract { held.mu.Unlock() held = q.heldSession(key) continue } held.stopTimer() if err == nil { break } q.log.Info("requester_held_session_policy_denied", "contract", contract, "err", err) if errors.Is(err, ErrPolicyRejected) { held.clear() q.releaseHeldSessionTarget(contract, privateKeyPath) } held.mu.Unlock() return commandFailed(io, "policy denied: ", err) } fresh := target == nil if target == nil { target, privateKeyPath, err = provision() if err != nil { held.mu.Unlock() return commandFailed(io, "provisioning failed: ", err) } contract = contractName held.name, held.vmName, held.privateKeyPath, held.target, held.contract = name, vmName, privateKeyPath, target, contract } ready := false if ticket, ticketErr := ValidateTicket(target.Address); ticketErr == nil { ready = provider.PollReady(ctx, privateKeyPath, ticket, DefaultVMReadyTimeout, target.HostKeys) } if !ready && reused { q.log.Warn("requester_held_session_lost", "contract", contract) held.clear() q.releaseHeldSessionTarget(contract, privateKeyPath) target, privateKeyPath, err = provision() if err != nil { held.mu.Unlock() return commandFailed(io, "provisioning failed: ", err) } contract = contractName held.name, held.vmName, held.privateKeyPath, held.target, held.contract = name, vmName, privateKeyPath, target, contract fresh = true ticket, ticketErr := ValidateTicket(target.Address) if ticketErr != nil { held.mu.Unlock() return commandFailed(io, "guest session failed: ", ticketErr) } ready = provider.PollReady(ctx, privateKeyPath, ticket, DefaultVMReadyTimeout, target.HostKeys) } if !ready { if fresh { held.clear() q.releaseHeldSessionTarget(contract, privateKeyPath) } held.mu.Unlock() return commandFailed(io, "guest session failed: ", errors.New("the guest ssh did not answer")) } ticket, err := ValidateTicket(target.Address) if err != nil { if fresh { held.clear() q.releaseHeldSessionTarget(contract, privateKeyPath) } held.mu.Unlock() return commandFailed(io, "guest session failed: ", err) } q.recordHeld(contract, name, privateKeyPath, q.cfg.SessionCap+q.cfg.SessionIdle) code := func() int { defer held.mu.Unlock() return provider.RunSession(ctx, privateKeyPath, ticket, renderExecCommand(command, lc), io, target.HostKeys) }() q.armHeldSession(key, held) io.Exit(code) return nil } func transientContractName(did, vmName string) string { sum := sha256.Sum256([]byte(did + "\x00" + vmName)) return transientContractPrefix + hex.EncodeToString(sum[:10]) } func (q *Requester) trackTransient(name string, cancel context.CancelFunc) func() { q.transientMu.Lock() if q.transient == nil { q.transient = map[string]context.CancelFunc{} } q.transient[name] = cancel q.transientWG.Add(1) q.transientMu.Unlock() return func() { q.transientMu.Lock() delete(q.transient, name) q.transientMu.Unlock() q.transientWG.Done() } } func (q *Requester) transientLive(name string) bool { q.transientMu.Lock() defer q.transientMu.Unlock() _, ok := q.transient[name] return ok } func (q *Requester) endTransientSessions(ctx context.Context, reason string) { q.transientMu.Lock() cancels := make([]context.CancelFunc, 0, len(q.transient)) for _, cancel := range q.transient { cancels = append(cancels, cancel) } q.transientMu.Unlock() if len(cancels) == 0 { return } q.log.Info("requester_transient_sessions_ended", "sessions", len(cancels), "reason", reason) for _, cancel := range cancels { cancel() } _ = supervise.Join(ctx, q.transientWG.Wait) } func (q *Requester) releaseStaleTransient() { started := q.startedAt.Load() for _, contract := range q.contracts.List() { switch { case !q.ownsContract(contract), contract.DeletionTimestamp != nil, contract.Spec.Release, contract.Labels[keepVMLabel] == "true", contract.Status.Phase == v1alpha1.ComputeContractReleased, contract.Status.Phase == v1alpha1.ComputeContractFailed, started != nil && !contract.CreationTimestamp.Time.Before(*started): continue case isHeldContractName(contract.Name): q.sweepHeldContract(contract) continue case !strings.HasPrefix(contract.Name, transientContractPrefix), q.transientLive(contract.Name): continue } q.log.Info("requester_stale_session_release", "contract", contract.Name, "phase", contract.Status.Phase, "reason", "no live ssh session holds this transient contract after a requester start") q.releaseHeldContract(contract.Name) } } func (q *Requester) contractState(name string) *v1alpha1.NodeClaimContract { contract, ok := q.contracts.Get(name) if !ok { return nil } return contract.Status.Contract } func heldContractName(did, name string) string { sum := sha256.Sum256([]byte(did + "\x00" + name)) return heldContractPrefix + hex.EncodeToString(sum[:10]) } func sessionRKey(label string, parts ...string) string { return market.SessionRKey(label, parts...) } func (q *Requester) existingSignedRecord(ctx context.Context, collection, rkey string) (market.SignedRecord, bool, error) { record, err := q.repo.GetRecord(ctx, q.repoDID, collection, rkey) if err != nil { if errors.Is(err, market.ErrNotFound) { return market.SignedRecord{}, false, nil } return market.SignedRecord{}, false, err } return market.SignedRecord{URI: record.URI, CID: record.CID, Record: record.Value}, true, nil } func contractTarget(contract v1alpha1.ComputeContract) (*SessionTarget, bool, error) { state := contract.Status.Contract if state == nil || state.VMAddress == "" || state.HostKeys == nil || len(state.HostKeys.Keys) == 0 || state.Receipt == nil || state.Receipt.URI == "" { return nil, false, nil } ticket, err := ValidateTicket(state.VMAddress) if err != nil { return nil, false, err } keys := make([]string, 0, len(state.HostKeys.Keys)) for _, key := range state.HostKeys.Keys { keys = append(keys, key.Type+" "+key.PublicKey) } return &SessionTarget{ VMName: state.VMName, Address: ticket, HostKeys: sshtransport.GuestHostKeys{Alias: hostKeyAlias(state.Receipt.URI), Keys: keys}, }, true, nil } func (q *Requester) awaitContractChange(ctx context.Context) error { calls := 0 return q.contracts.Await(ctx, func() bool { calls++ return calls > 1 }) } func (q *Requester) heldSessionAllowed(ctx context.Context, contract, policyName string) error { held, ok := q.contracts.Get(contract) if !ok || held.Status.Contract == nil { return errors.New("the held session contract " + contract + " is not known") } return q.recordedWinnerAllowed(ctx, held.Status.Contract, policyName) } func (q *Requester) provisionHeldContract(ctx context.Context, name, vmName, did, publicKey, policyName string, policyArgs map[string]string, userData string, ttlSeconds int32, labels map[string]string) (*SessionTarget, error) { args := "" if len(policyArgs) > 0 { raw, err := json.Marshal(policyArgs) if err != nil { return nil, err } args = string(raw) } deadline := q.now().Add(q.cfg.BidWindow + q.cfg.ReceiptTimeout + DefaultVMReadyTimeout) waitCtx, cancel := context.WithDeadline(ctx, deadline) defer cancel() contract := v1alpha1.ComputeContract{ TypeMeta: metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: contractmarket.KindComputeContract}, ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels}, Spec: v1alpha1.ComputeContractSpec{ ComputeContractTemplate: v1alpha1.ComputeContractTemplate{VMName: vmName, SshAuthorizedKeys: []string{publicKey}, Policy: policyName, PolicyArgs: args, UserData: userData}, OwnerDid: did, TTLSecondsAfterFinished: new(ttlSeconds), }, } for { if existing, ok := q.contracts.Get(name); ok { if existing.DeletionTimestamp == nil { if err := q.deps.Intents.DeleteComputeContract(ctx, name); err != nil { return nil, err } } gone := func() bool { _, present := q.contracts.Get(name) return !present } if err := q.contracts.Await(waitCtx, gone); err != nil { if ctx.Err() != nil { return nil, ctx.Err() } return nil, errors.New("the previous session contract was not removed in time") } continue } err := q.deps.Intents.CreateComputeContract(ctx, contract) if err == nil { break } if !errors.Is(err, contractmarket.ErrExists) { return nil, err } if err := q.awaitContractChange(waitCtx); err != nil { if ctx.Err() != nil { return nil, ctx.Err() } return nil, errors.New("the session contract was not created in time") } } return q.waitHeldContract(waitCtx, ctx, name) } func (q *Requester) waitHeldContract(waitCtx, ctx context.Context, name string) (*SessionTarget, error) { var target *SessionTarget var failure error seen := false _ = q.contracts.Await(waitCtx, func() bool { contract, ok := q.contracts.Get(name) if !ok { if seen { failure = errors.New("the session contract disappeared") } return seen } seen = true switch contract.Status.Phase { case v1alpha1.ComputeContractFailed: failure = fmt.Errorf("the session contract failed: %s", contract.Status.Message) return true case v1alpha1.ComputeContractReleased: failure = errors.New("the session contract was released before the guest was ready") return true } found, ready, err := contractTarget(contract) if err != nil { failure = err return true } target = found return ready }) switch { case failure != nil: return nil, failure case target != nil: return target, nil case ctx.Err() != nil: return nil, ctx.Err() } return nil, errors.New("the guest did not report onNetwork and hostKeys in time") } func (q *Requester) releaseHeldContract(name string) { if name == "" { return } for attempt := 1; ; attempt++ { ctx, cancel := context.WithTimeout(context.WithoutCancel(q.runContext()), DefaultReleaseTimeout) err := q.deps.Intents.ReleaseComputeContract(ctx, name) cancel() if err == nil || errors.Is(err, contractmarket.ErrNotFound) { if attempt > 1 { q.log.Info("requester_held_contract_released", "contract", name, "attempt", attempt) } return } q.log.Warn("requester_held_contract_release_failed", "contract", name, "attempt", attempt, "err", err) if attempt >= heldReleaseAttempts { q.log.Warn("requester_held_contract_release_left", "contract", name, "reason", "the next requester start releases it") return } time.Sleep(min(time.Second<<(attempt-1), heldReleaseBackoffMax)) } } func (q *Requester) releaseHeldSessionTarget(contract, privateKeyPath string) { if contract != "" { q.releaseHeldContract(contract) } _ = sshtransport.RemoveGuestKeypair(privateKeyPath) } func (q *Requester) sessionBestBid(ctx context.Context, rfp market.SignedRecord, bids map[string]BidRecord, workflow PolicyGate, full map[string]bool, limit int) *BidRecord { fullKey := func(bid BidRecord) string { return strings.Join([]string{bid.URI, bid.CID, workflow.Key(), rfp.URI, rfp.CID}, "|") } var best *BidRecord for denied := 0; ; { if limit > 0 && denied >= limit { return nil } best = nil for _, bid := range bids { candidate := bid if err := q.classifyBid(ctx, &candidate); err != nil { continue } if allowed, done := full[fullKey(candidate)]; done && !allowed { continue } if !q.bidAllowedWithPolicy(ctx, candidate, workflow) { continue } switch { case best == nil: best = &candidate case best.Free: case candidate.Free: best = &candidate case candidate.Cost < best.Cost: best = &candidate } } if best == nil { return nil } key := fullKey(*best) if _, done := full[key]; !done { decision := q.policy.Evaluate(ctx, workflow.Input(PolicyInput{ SelfDid: q.marketDID, SubjectDid: best.DID, CounterpartyDid: best.DID, RootRequesterDid: q.marketDID, RFP: &PolicyRef{URI: rfp.URI, CID: rfp.CID}, Bid: &PolicyRef{URI: best.URI, CID: best.CID}, })) full[key] = decision.Allow if !decision.Allow { denied++ q.log.Info("requester_bid_policy_denied", "bidder", best.DID, "bid", best.URI, "mode", participant.PolicyModeFull, "reason", decision.Reason) } } if full[key] { break } } return best } func lcEnv(env map[string]string) map[string]string { out := map[string]string{} for key, value := range env { if !strings.HasPrefix(key, "LC_") || localeNames[key] { continue } out[key] = value } return out } func renderExecCommand(command string, env map[string]string) string { if len(env) == 0 { return command } keys := make([]string, 0, len(env)) for key := range env { keys = append(keys, key) } slices.Sort(keys) var b strings.Builder for _, key := range keys { b.WriteString("export ") b.WriteString(key) b.WriteString("=") b.WriteString("'" + strings.ReplaceAll(env[key], "'", "'\"'\"'") + "'") b.WriteString("; ") } b.WriteString(command) return b.String() } func truthy(value string) bool { switch strings.ToLower(strings.TrimSpace(value)) { case "1", "true", "yes", "on": return true default: return false } }