package main import ( "cmp" "context" "encoding/json" "errors" "fmt" "io" "log/slog" "os" "os/exec" "path/filepath" "slices" "strconv" "strings" "sync" "time" "k8s.io/apimachinery/pkg/api/meta" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose" "github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketnetwork" "github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketrequester" "github.com/johnandersen777/socialweb-computer-kcp/internal/appcli" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/cloudinit" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market" "github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap" "github.com/johnandersen777/socialweb-computer-kcp/internal/clicfg" "github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/logx" "github.com/johnandersen777/socialweb-computer-kcp/internal/market/requester" "github.com/johnandersen777/socialweb-computer-kcp/internal/marketidentity/atproto" "github.com/johnandersen777/socialweb-computer-kcp/internal/stack/request" "github.com/johnandersen777/socialweb-computer-kcp/internal/webrtcproxy" ) const ( requestSSHReady = time.Minute requestIdentityWait = 5 * time.Minute requestSummaryRead = 10 * time.Second requestAttachProbe = 10 * time.Second requestOwnerExitWait = 5 * time.Minute requestSSHRetryRun = 20 * time.Second transportIroh = "iroh" transportWebRTC = "webrtc" sshPort = 22 sshConnectionFailed = 255 ) var requestOptions = []string{"network", "name", "secrets", "tunnel", "agent", "guest-tools", "association-ref", "hold", "exec", "skip-ssh", "contract-timeout", "state-dir", "ready-timeout", "network-timeout", "kcp-path", "kine-path", "bao-path", "dumbpipe-path", "asset-cache", "transport", "tag", "log-level"} var requestCreateOptions = append(append([]string{}, requestOptions...), "identity", "associate", "guest-requester") type requestRun struct { a *app vals clicfg.Values stateRoot string inv request.Invocation hold bool skipSSH bool exec string secrets string associate string associationRefArg string transport string tag string target string root v1alpha1.MarketRequesterSpec contract v1alpha1.ComputeContractSpec appOptions []v1alpha1.ApplicationOption contractTimeout time.Duration networkTimeout time.Duration summary request.Summary kcp *request.Kcp standalone devinstance.Layout instanceDir string resuming bool cleanupOnly bool identity *request.IdentityStore identityState *request.Identity identityObject *v1alpha1.MarketIdentity expectedPDSURL string awaitRefFile string guestRequester bool handshakeStarted bool homeDir string verb string kubeconfig string workspace string } func (r *requestRun) home() string { return cmp.Or(r.homeDir, request.Dir(r.stateRoot, r.inv.Name)) } func (r *requestRun) openIdentity(name string) error { idName := strings.TrimSpace(r.vals.String("identity")) if idName == "" { return nil } if r.inv.Mode == request.ModeNetwork { return errors.New("--identity is for a standalone invocation; a market network places its child before its state dir exists") } if r.root.Identity.Mode != "" && r.root.Identity.Mode != v1alpha1.MarketIdentityModeLocalRepo { return fmt.Errorf("--identity needs the localRepo identity mode, not %s", r.root.Identity.Mode) } if r.root.Identity.PdsURL != "" { return errors.New("--identity does not take --atproto-pds-url") } store, err := request.OpenIdentity(r.stateRoot, idName) if err != nil { return err } fail := func(err error) error { store.Close() return err } id, err := store.Load() if err != nil { return fail(err) } if held, err := store.PDSLockHeld(); err != nil { return fail(err) } else if held { return fail(fmt.Errorf("the identity %s is in use by a running PDS; release or stop its invocation first", idName)) } if id != nil { if id.DID == "" { accounts, err := request.ReadAccounts(store.PDSDir()) if err != nil { return fail(err) } switch len(accounts) { case 1: id.DID, id.Handle = accounts[0].DID, accounts[0].Handle case 0: default: return fail(fmt.Errorf("the identity %s PDS holds %d accounts; one identity serves one account", idName, len(accounts))) } } if id.DID == "" && id.PDSPublicURL != "" { return fail(fmt.Errorf("the identity %s names a PDS URL but no DID; the store is partial", idName)) } if id.Handle != "" && r.root.Identity.Handle != "" && !strings.EqualFold(r.root.Identity.Handle, id.Handle) { return fail(fmt.Errorf("the identity %s uses handle %s, not %s", idName, id.Handle, r.root.Identity.Handle)) } r.root.Identity.Mode = v1alpha1.MarketIdentityModeLocalRepo r.root.Identity.Did = id.DID if id.Handle != "" { r.root.Identity.Handle = id.Handle } switch v1alpha1.AssociationMode(id.Association) { case v1alpha1.AssociationSkip: r.root.Identity.Association = v1alpha1.AssociationSkip case v1alpha1.AssociationRequireApproval: r.root.Identity.Association = v1alpha1.AssociationRequireApproval default: return fail(fmt.Errorf("the identity %s has association %q", idName, id.Association)) } r.root.Identity.ApprovedAssociations = append([]string(nil), id.ApprovedAssociations...) if r.root.Identity.Association == v1alpha1.AssociationSkip && (r.root.Identity.AssociationRef != nil || r.awaitRefFile != "") { return fail(fmt.Errorf("the identity %s has association Skip; --association-ref needs a RequireApproval identity", idName)) } if r.root.Identity.Association == v1alpha1.AssociationRequireApproval && len(r.root.Identity.ApprovedAssociations) == 0 && r.root.Identity.AssociationRef == nil && r.awaitRefFile == "" { return fail(fmt.Errorf("the identity %s is RequireApproval with no approved operator; pass --association-ref at:////#, or use --skip-qr to create a skip identity", idName)) } if id.PLCDirectoryURL != "" && r.root.Network != nil && r.root.Network.PLCDirectoryURL != "" && !strings.EqualFold(strings.TrimRight(id.PLCDirectoryURL, "/"), strings.TrimRight(r.root.Network.PLCDirectoryURL, "/")) { return fail(fmt.Errorf("the identity %s uses PLC %s, not %s", idName, id.PLCDirectoryURL, r.root.Network.PLCDirectoryURL)) } if id.DID != "" { rot, err := store.Validate(id.DID, id.PDSPublicURL != "") if err != nil { return fail(fmt.Errorf("the identity %s: %w", idName, err)) } if id.RotationKeyDID != "" && rot != id.RotationKeyDID { return fail(fmt.Errorf("the identity %s rotation key is %s, not the recorded %s", idName, rot, id.RotationKeyDID)) } id.RotationKeyDID = rot } r.identityState = id } else { if entries, err := os.ReadDir(store.PDSDir()); err == nil && len(entries) > 0 { return fail(fmt.Errorf("the identity %s has PDS state but no identity.json; refusing to adopt it", idName)) } else if err != nil && !errors.Is(err, os.ErrNotExist) { return fail(err) } association := v1alpha1.AssociationSkip if r.root.Identity.AssociationRef != nil || r.awaitRefFile != "" { association = v1alpha1.AssociationRequireApproval } else if r.root.Identity.Association != v1alpha1.AssociationSkip { return fail(fmt.Errorf("a new identity needs -- --skip-qr or --association-ref at:////#")) } r.root.Identity.Association = association r.identityState = &request.Identity{Schema: request.IdentitySchema, Name: idName, Mode: "localRepo", Association: string(association)} } if err := store.EnsureDirs(); err != nil { return fail(err) } r.identityState.LastInvocation = &request.IdentityInvocation{Name: name, StateDir: request.InstanceStateDirIn(r.home(), name)} if err := store.Save(r.identityState); err != nil { return fail(err) } r.expectedPDSURL = strings.TrimSpace(r.identityState.PDSPublicURL) r.appOptions = append(r.appOptions, v1alpha1.ApplicationOption{Name: "pds-state-dir", Value: store.PDSDir()}) if r.expectedPDSURL != "" { r.appOptions = append(r.appOptions, v1alpha1.ApplicationOption{Name: "expected-pds-url", Value: r.expectedPDSURL}) } r.root.Options = r.appOptions r.identity = store r.inv.Identity = idName r.summary.Identity = idName r.log("the identity is "+idName, "did", r.identityState.DID, "store", store.Dir()) return nil } func (r *requestRun) linkIdentityExposure() error { if r.identity == nil { return nil } return r.identity.LinkExposure(request.InstanceStateDirIn(r.home(), r.inv.Name)) } func (r *requestRun) saveIdentityDID(obj *v1alpha1.MarketIdentity) { if r.identity == nil || r.identityState == nil || obj == nil { return } did := strings.TrimSpace(obj.Status.Did) if did == "" { return } handle := strings.TrimSpace(obj.Status.Handle) if r.identityState.DID == did && (handle == "" || r.identityState.Handle == handle) { return } r.identityState.DID = did if handle != "" { r.identityState.Handle = handle } if err := r.identity.Save(r.identityState); err != nil { r.a.log.Warn("writing the identity DID", "identity", r.identity.Name(), "err", err.Error()) } } func (r *requestRun) finalizeIdentity() error { if r.identity == nil || r.identityState == nil { return nil } obj := r.identityObject if obj == nil { ids, err := r.kcp.Identities() if err != nil { return err } ctx, cancel := context.WithTimeout(context.Background(), requestSummaryRead) defer cancel() obj, err = ids.GetMarketIdentity(ctx, marketrequester.RequesterIdentity) if err != nil { return err } } did := strings.TrimSpace(obj.Status.Did) if did == "" { return errors.New("the requester identity reports no DID") } if r.identityState.DID != "" && r.identityState.DID != did { return fmt.Errorf("the requester DID is %s, not the identity's %s", did, r.identityState.DID) } r.identityState.DID = did r.identityState.Mode = "localRepo" if handle := strings.TrimSpace(obj.Status.Handle); handle != "" { r.identityState.Handle = handle } if pdsURL := strings.TrimSpace(obj.Status.PdsURL); pdsURL != "" { r.identityState.PDSPublicURL = pdsURL } r.identityState.Association = string(r.root.Identity.Association) r.identityState.ApprovedAssociations = append([]string(nil), r.root.Identity.ApprovedAssociations...) if r.root.Network != nil { r.identityState.PLCDirectoryURL = r.root.Network.PLCDirectoryURL } if rot, err := r.identity.ReadRotationKeyDID(); err == nil { r.identityState.RotationKeyDID = rot } if r.identityState.LastInvocation == nil { r.identityState.LastInvocation = &request.IdentityInvocation{Name: r.inv.Name, StateDir: request.InstanceStateDirIn(r.home(), r.inv.Name)} } return r.identity.Save(r.identityState) } func splitWords(rest []string) ([]string, []string) { i := 0 for i < len(rest) && !strings.HasPrefix(rest[i], "--") { i++ } pass := rest[i:] if len(pass) > 0 && pass[0] == "--" { pass = pass[1:] } return rest[:i], pass } func (r *requestRun) log(msg string, args ...any) { r.a.log.Info(msg, append([]any{"request", r.inv.Name, "utc", time.Now().UTC().Format(time.RFC3339)}, args...)...) } func newRequestRun(a *app, vals clicfg.Values) (*requestRun, error) { root, err := devinstance.StateRoot(vals.String("state-dir"), a.env) if err != nil { return nil, err } if root, err = filepath.Abs(root); err != nil { return nil, err } r := &requestRun{a: a, vals: vals, stateRoot: root, hold: vals.Bool("hold"), skipSSH: vals.Bool("skip-ssh"), exec: vals.String("exec"), secrets: vals.String("secrets"), tag: strings.TrimSpace(vals.String("tag")), target: strings.TrimSpace(vals.String("contract"))} switch r.transport = strings.ToLower(strings.TrimSpace(vals.String("transport"))); r.transport { case "", transportIroh: r.transport = transportIroh case transportWebRTC: if r.tag == "" { return nil, errors.New("transport webrtc: --tag names the webrtc tunnel") } default: return nil, fmt.Errorf("transport %q: want iroh or webrtc", r.transport) } if r.contract.Tunnels, err = parseTunnels(vals.String("tunnel")); err != nil { return nil, fmt.Errorf("tunnel: %w", err) } r.contract.Agent = agentOf(vals.String("agent")) if r.contract.GuestTools, err = guestToolsOf(vals.String("guest-tools")); err != nil { return nil, fmt.Errorf("guest-tools: %w", err) } if r.contractTimeout, err = time.ParseDuration(vals.String("contract-timeout")); err != nil { return nil, fmt.Errorf("contract-timeout: %w", err) } if r.networkTimeout, err = time.ParseDuration(vals.String("network-timeout")); err != nil { return nil, fmt.Errorf("network-timeout: %w", err) } return r, nil } func splitList(v string) []string { var out []string for s := range strings.SplitSeq(v, ",") { if s = strings.TrimSpace(s); s != "" { out = append(out, s) } } return out } func (r *requestRun) passthrough(args []string, network bool) error { root, err := parseRequesterSpec(args, "run request", requestCreateOptions, network, r.contractArg) if err != nil { return err } r.root, r.appOptions = root, root.Options r.contract.Policy = cmp.Or(strings.TrimSpace(r.contract.Policy), requester.DefaultPolicyWorkflow) return nil } func (r *requestRun) contractArg(p compose.AppArg) (bool, error) { switch { case p.To == compose.MapToCLI: switch p.Name { case "exec": r.exec = p.Value case "hold", "keep-vm": r.hold = p.Value != "false" case "skip-ssh": r.skipSSH = p.Value != "false" } case p.To == "contract.bidderDids": r.contract.BidderDids = append(r.contract.BidderDids, splitList(p.Value)...) case p.To == "contract.denyBidderDids": r.contract.DenyBidderDids = append(r.contract.DenyBidderDids, splitList(p.Value)...) case p.To == "contract.vmName": r.contract.VMName = p.Value case p.To == "contract.computeKind": switch kind := v1alpha1.ComputeKind(strings.TrimSpace(p.Value)); kind { case v1alpha1.ComputeKindVM, v1alpha1.ComputeKindDeno: r.contract.ComputeKind = kind default: return true, fmt.Errorf("--compute-kind %q: want vm or deno", p.Value) } case p.To == "contract.userData": b, err := os.ReadFile(p.Value) if err != nil { return true, fmt.Errorf("--user-data: %w", err) } r.contract.UserData = string(b) case p.To == "contract.sshAuthorizedKeys": r.contract.SshAuthorizedKeys = append(r.contract.SshAuthorizedKeys, splitList(p.Value)...) case p.To == "contract.policy": r.contract.Policy = p.Value case p.To == "contract.policyArgs": r.contract.PolicyArgs = p.Value case p.To == "contract.vmReadyTimeoutSec": n, err := strconv.Atoi(p.Value) if err != nil { return true, fmt.Errorf("--vm-ready-timeout-sec %q: %w", p.Value, err) } r.contract.VMReadyTimeoutSec = n case p.To == "contract.secretsRef": r.secrets = p.Value case p.To == "contract.agent": r.contract.Agent = agentOf(p.Value) case p.To == "contract.association": r.associate = p.Value case p.To == "spec.identity.associationRef": r.associationRefArg = p.Value case p.To == "contract.guestTools": tools, err := guestToolsOf(p.Value) if err != nil { return true, fmt.Errorf("--guest-tools: %w", err) } for _, tool := range tools { if !slices.Contains(r.contract.GuestTools, tool) { r.contract.GuestTools = append(r.contract.GuestTools, tool) } } case p.To == "contract.tunnels": tunnels, err := parseTunnels(p.Value) if err != nil { return true, fmt.Errorf("--tunnel: %w", err) } r.contract.Tunnels = append(r.contract.Tunnels, tunnels...) default: return false, nil } return true, nil } func parseTunnels(value string) ([]v1alpha1.ComputeContractTunnel, error) { var out []v1alpha1.ComputeContractTunnel for _, item := range splitList(value) { parts := strings.Split(item, ":") if len(parts) < 2 || len(parts) > 4 { return nil, fmt.Errorf("%q: want :[:][:shell[=]] or ::", item) } port, err := strconv.Atoi(parts[1]) if err != nil { return nil, fmt.Errorf("%q: port: %w", item, err) } tunnel := v1alpha1.ComputeContractTunnel{Tag: parts[0], Port: port} rest := parts[2:] if len(rest) > 0 && !strings.HasPrefix(rest[0], "/") && !isShellForm(rest[0]) { switch transport := v1alpha1.TunnelTransport(rest[0]); transport { case v1alpha1.TunnelTransportOpenTunnel, v1alpha1.TunnelTransportWebRTC: tunnel.Transport = transport default: return nil, fmt.Errorf("%q: transport %q is not %s or %s", item, rest[0], v1alpha1.TunnelTransportOpenTunnel, v1alpha1.TunnelTransportWebRTC) } rest = rest[1:] } switch { case len(rest) == 0: case len(rest) == 1 && strings.HasPrefix(rest[0], "/") && len(parts) == 3: tunnel.Shell = &v1alpha1.ComputeContractShell{EnvFile: rest[0]} case len(rest) == 1 && isShellForm(rest[0]): envFile, _ := strings.CutPrefix(strings.TrimPrefix(rest[0], "shell"), "=") tunnel.Shell = &v1alpha1.ComputeContractShell{EnvFile: envFile} default: return nil, fmt.Errorf("%q: want :[:][:shell[=]] or ::", item) } if tunnel.Shell != nil && tunnel.Transport == v1alpha1.TunnelTransportWebRTC { return nil, fmt.Errorf("%q: a wootty shell is served over opentunnel, not webrtc", item) } out = append(out, tunnel) } return out, nil } func isShellForm(part string) bool { return part == "shell" || strings.HasPrefix(part, "shell=") } func (r *requestRun) webrtcTunnel() error { if r.transport != transportWebRTC { return nil } for _, t := range r.contract.Tunnels { if t.Tag == r.tag { if t.Transport != v1alpha1.TunnelTransportWebRTC { return fmt.Errorf("--transport webrtc: the tunnel %s has transport %s, not webrtc", t.Tag, cmp.Or(string(t.Transport), string(v1alpha1.TunnelTransportOpenTunnel))) } return nil } } r.contract.Tunnels = append(r.contract.Tunnels, v1alpha1.ComputeContractTunnel{Tag: r.tag, Port: sshPort, Transport: v1alpha1.TunnelTransportWebRTC}) return nil } func secretsApplied(c *v1alpha1.ComputeContract) bool { return c.Status.SecretsFetchedAt != "" || meta.IsStatusConditionTrue(c.Status.Conditions, v1alpha1.ComputeContractConditionSecretsApplied) } func readSecrets(path string) ([]byte, error) { b, err := os.ReadFile(path) if err != nil { return nil, err } var entries []struct { Path string `json:"path"` Value any `json:"value"` } if err := json.Unmarshal(b, &entries); err != nil { return nil, fmt.Errorf("%s is not a JSON array of {path, value}: %w", path, err) } for i, e := range entries { if !strings.HasPrefix(e.Path, "/") || strings.Contains(e.Path, "/../") { return nil, fmt.Errorf("%s: entry %d path %q is not an absolute path", path, i, e.Path) } } return b, nil } func (r *requestRun) networkLayout() (devinstance.Layout, devinstance.Record, error) { l, err := r.a.layoutOf(r.inv.Network, r.vals) if err != nil { return l, devinstance.Record{}, err } var rec devinstance.Record if err := fsx.ReadJSON(l.Record, &rec); err != nil { return l, rec, fmt.Errorf("market network %s has no instance record (dev up --profile market-network first): %w", r.inv.Network, err) } if rec.Profile != bootstrap.ProfileMarketNetwork || rec.AdminKubeconfig == "" { return l, rec, fmt.Errorf("instance %s is not a market network (profile %s)", r.inv.Network, rec.Profile) } return l, rec, nil } func (r *requestRun) childRecord(netLayout devinstance.Layout, instance string) (devinstance.Record, error) { cl, err := devinstance.NewLayout(instance, filepath.Join(netLayout.StateDir, compose.AppsDir), filepath.Join(netLayout.RuntimeDir, compose.AppsDir)) if err != nil { return devinstance.Record{}, err } var rec devinstance.Record if err := fsx.ReadJSON(cl.Record, &rec); err != nil { return rec, err } return rec, nil } func (r *requestRun) start(ctx context.Context, fresh bool) error { started := time.Now() var kubeconfig, workspace, netKubeconfig string switch r.inv.Mode { case request.ModeNetwork: netLayout, netRec, err := r.networkLayout() if err != nil { return err } netKubeconfig = netRec.AdminKubeconfig if fresh { app := v1alpha1.NetworkApp{Name: r.inv.Name, Kind: v1alpha1.ApplicationKindMarketRequester, Profile: marketrequester.RequesterProfileNetwork, Options: r.appOptions, SecretsFallback: r.root.SecretsFallback, RequesterIngress: r.root.RequesterIngress} if r.root.Comms != nil { app.Components = v1alpha1.ComponentOverrides{compose.ComponentKindRequester: {Comms: r.root.Comms}} } if err := request.AddToNetwork(ctx, netRec.AdminKubeconfig, app); err != nil { return err } } else if err := request.SetStopped(ctx, netRec.AdminKubeconfig, r.inv.Name, false); err != nil { return err } r.log("the market network places the invocation in its workspace of the network's kcp", "network", r.inv.Network) app, err := request.WaitNetworkApp(ctx, netRec.AdminKubeconfig, r.inv.Name, r.networkTimeout, func(a v1alpha1.NetworkAppStatus) bool { return a.Instance != "" && a.KcpURL != "" && a.Message != marketnetwork.StoppedMessage && a.Phase != v1alpha1.ApplicationPhasePending }, func(a v1alpha1.NetworkAppStatus) { r.log("network child", "phase", string(a.Phase), "message", a.Message) }) if err != nil { return err } rec, err := r.childRecord(netLayout, app.Instance) if err != nil { return err } r.instanceDir = rec.Layout.StateDir r.inv.Instance, r.summary.Instance, r.summary.KcpURL = app.Instance, app.Instance, rec.KcpURL kubeconfig, workspace = rec.AdminKubeconfig, rec.BaseWorkspace default: vals := r.vals.With("instance", r.inv.Name).With("profile", bootstrap.ProfileQuick). With("state-dir", r.home()).With(appRoleOpt, devinstance.RoleRequester) up, code := bringUp(ctx, r.a, vals, false) if code != exitOK { return fmt.Errorf("the invocation's kcp did not come up (exit %d)", code) } r.standalone, r.instanceDir = up.layout, up.layout.StateDir r.inv.Instance, r.summary.Instance, r.summary.KcpURL = up.Instance, up.Instance, up.KcpURL kubeconfig = up.AdminKubeconfig } k, err := request.OpenKcp(kubeconfig, workspace) if err != nil { return err } r.kubeconfig, r.workspace = kubeconfig, workspace r.kcp = k.ForContract(r.target) if r.inv.Mode == request.ModeStandalone && fresh { if err := k.ApplyRoot(ctx, r.root); err != nil { return fmt.Errorf("applying the MarketRequester root object: %w", err) } } r.summary.KcpColdSeconds = time.Since(started).Seconds() r.summary.KcpRunning = true r.log("the invocation's kcp runs", "kcp", r.summary.KcpURL, "workspace", workspace, "instance", r.inv.Instance, "coldSeconds", r.summary.KcpColdSeconds) if err := r.signIn(ctx, fresh || r.resuming); err != nil { return fmt.Errorf("signing in: %w", err) } if err := r.kcp.WaitReady(ctx, r.networkTimeout, func(st v1alpha1.ApplicationStatus) { r.log("requester", "phase", string(st.Phase), "message", st.Message, "components", len(st.Components)) }); err != nil { return err } if netKubeconfig != "" { if err := request.WaitNetworkVouches(ctx, netKubeconfig, r.inv.Name, r.networkTimeout, func(ws, record string) { r.log("waiting for the network's vouch", "workspace", ws, "record", record) }); err != nil { return err } } return request.WriteInvocationIn(r.home(), r.inv) } func (r *requestRun) stop(ctx context.Context, remove bool) { r.keepLogs() switch r.inv.Mode { case request.ModeNetwork: _, netRec, err := r.networkLayout() if err != nil { r.a.log.Warn("stopping the invocation", "err", err.Error()) return } if remove { err = request.RemoveFromNetwork(ctx, netRec.AdminKubeconfig, r.inv.Name) } else { err = request.SetStopped(ctx, netRec.AdminKubeconfig, r.inv.Name, true) } if err != nil { r.a.log.Warn("stopping the invocation", "err", err.Error()) return } if !remove { _, err = request.WaitNetworkApp(ctx, netRec.AdminKubeconfig, r.inv.Name, r.networkTimeout, func(a v1alpha1.NetworkAppStatus) bool { return a.Message == marketnetwork.StoppedMessage }, nil) } else { err = request.WaitNetworkGone(ctx, netRec.AdminKubeconfig, r.inv.Name, r.networkTimeout) } if err != nil { r.a.log.Warn("stopping the invocation", "err", err.Error()) } default: l := r.standalone if l.ID == "" { vals := r.vals.With("instance", r.inv.Name).With("state-dir", r.home()) var err error if l, err = r.a.layoutOf(r.inv.Name, vals); err != nil { r.a.log.Warn("stopping the invocation", "err", err.Error()) return } } d := downInstance(ctx, r.a, l, !remove, true) if len(d.Remaining) > 0 { r.a.log.Warn("processes of the invocation's kcp remain", "pids", fmt.Sprint(d.Remaining)) } } r.summary.KcpRunning = false r.log("the invocation's kcp is stopped", "stateKept", !remove) } func (r *requestRun) writeSummary(ctx context.Context, c *v1alpha1.ComputeContract) { r.summary.UpdatedAt = time.Now().UTC() if c != nil { r.summary.SecretsFetchedAt = c.Status.SecretsFetchedAt r.summary.Phase, r.summary.Message, r.summary.SecretsUrl = string(c.Status.Phase), c.Status.Message, c.Status.SecretsURL r.summary.Contract = request.SummaryOf(c) } ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), requestSummaryRead) defer cancel() if st, err := r.kcp.Root(ctx); err == nil && st.Did != "" { r.summary.Did = st.Did } if err := request.WriteSummary(r.stateRoot, r.summary); err != nil { r.a.log.Warn("writing summary.json", "err", err.Error()) } } func (r *requestRun) putPdsAccountPassword(ctx context.Context) error { if r.root.Identity.Mode != v1alpha1.MarketIdentityModePdsAccount { return nil } password := strings.TrimSpace(os.Getenv(compose.EnvPdsAccountPassword)) if password == "" { return nil } ref, err := r.kcp.PutPdsAccountPassword(ctx, password, r.networkTimeout) if err != nil { return fmt.Errorf("putting %s into the requester's own OpenBao: %w", compose.EnvPdsAccountPassword, err) } r.log("the pdsAccount password is in the requester's own OpenBao", "mount", ref.Mount, "path", ref.Path) return nil } func (r *requestRun) signIn(ctx context.Context, untilAssociated bool) error { if err := r.putPdsAccountPassword(ctx); err != nil { return err } if err := r.kcp.WaitIdentityObject(ctx, r.networkTimeout); err != nil { return err } ids, err := r.kcp.Identities() if err != nil { return err } var in io.Reader if fi, err := os.Stdin.Stat(); err == nil && fi.Mode()&os.ModeCharDevice != 0 { in = os.Stdin } wctx, cancel := context.WithCancelCause(ctx) defer cancel(nil) deadline := time.AfterFunc(requestIdentityWait, func() { cancel(fmt.Errorf("the requester identity asked for no sign-in step and was not Ready within %s", requestIdentityWait)) }) defer deadline.Stop() watch := func(ctx context.Context, name string, each func(*v1alpha1.MarketIdentity) (bool, error)) (*v1alpha1.MarketIdentity, error) { return r.kcp.AwaitIdentity(ctx, name, func(obj *v1alpha1.MarketIdentity) (bool, error) { r.saveIdentityDID(obj) if !untilAssociated && obj != nil && obj.Status.Phase == v1alpha1.MarketIdentityPhaseAwaitingApproval { return true, nil } if err := associationRefRefused(obj); err != nil { return false, err } if err := r.guestHandshake(ctx, obj); err != nil { return false, err } if waitsOnPerson(obj) && deadline.Stop() { r.log(cmp.Or(r.verb, "run request")+": the requester identity waits on a person (a QR scan, a browser sign-in or the operator's association); waiting without a deadline until it is Ready, Ctrl-C stops and keeps the invocation's state", "identity", name, "phase", string(obj.Status.Phase)) } return each(obj) }) } id, err := appcli.IdentityRenderer{Client: ids, Watch: watch, Out: r.a.stderr, In: in, Hint: r.identityHint}.WaitReady(wctx, marketrequester.RequesterIdentity) if id != nil { r.identityObject = id } if cause := context.Cause(wctx); err != nil && cause != nil && ctx.Err() == nil { return fmt.Errorf("%w: %w", cause, err) } return err } func waitsOnPerson(obj *v1alpha1.MarketIdentity) bool { if obj == nil { return false } switch obj.Status.Phase { case v1alpha1.MarketIdentityPhaseReady, v1alpha1.MarketIdentityPhaseFailed: return false } if obj.Status.PendingAction != nil { return true } switch obj.Status.Phase { case v1alpha1.MarketIdentityPhaseAwaitingScan, v1alpha1.MarketIdentityPhaseAwaitingAuthorization, v1alpha1.MarketIdentityPhaseAwaitingApproval, v1alpha1.MarketIdentityPhaseSessionExpired: return true } return false } func (r *requestRun) identityHint(obj *v1alpha1.MarketIdentity) string { if c := meta.FindStatusCondition(obj.Status.Conditions, atproto.ConditionReady); c != nil && c.Reason == atproto.ReasonWaitingForCredential && obj.Spec.Mode == v1alpha1.MarketIdentityModePdsAccount { hint := "The pdsAccount identity waits for its app password in the requester's own OpenBao. Give it " + compose.EnvPdsAccountPassword + " in this command's environment" if r.verb == requesterVerb { hint += ", or, while the requester runs: socialweb-computer run requester password " + r.inv.Name + " < " } return hint } if obj.Status.Phase != v1alpha1.MarketIdentityPhaseAwaitingApproval || obj.Status.PendingAction != nil { return "" } did := cmp.Or(obj.Status.Did, "") if obj.Spec.AssociationRef != nil { return associationRefHint(obj, did) } return strings.Join([]string{ "The requester's DID " + did + " is a new localRepo DID with association RequireApproval. No bidder accepts it", "until the operator's verified association claim names it. The operator writes that claim from a checkout", "of socialweb-computer-kcp, with the operator account's credential file:", " deno run -A --no-lock scripts/associate-bidder.ts --creds --account --requester-did " + did, "This invocation keeps waiting for it; Ctrl-C stops it before any contract is posted.", "The prod CLI requester signs in with an existing account instead: socialweb-computer run request ... -- --atproto-oauth-qr", }, "\n") } func leftHeld(c *v1alpha1.ComputeContract) bool { return c.Status.Phase != v1alpha1.ComputeContractHeld } func contractSettled(c *v1alpha1.ComputeContract) bool { switch c.Status.Phase { case v1alpha1.ComputeContractHeld, v1alpha1.ComputeContractFailed, v1alpha1.ComputeContractReleased: return c.Status.ObservedGeneration == c.Generation || c.Status.ObservedGeneration == 0 } return false } func (r *requestRun) progress(c *v1alpha1.ComputeContract) { phase := "" if c.Status.Contract != nil { phase = string(c.Status.Contract.Phase) } r.log("contract", "phase", string(c.Status.Phase), "contractPhase", phase, "message", c.Status.Message) } func (r *requestRun) sshClient(ctx context.Context) (request.SSH, error) { s := request.SSH{Dir: filepath.Join(r.home(), request.SSHDir)} if r.transport == transportWebRTC { proxy, err := r.proxyCommand() if err != nil { return s, err } s.ProxyCommand = proxy return s, nil } dp, err := r.a.devExe(ctx, r.vals, "dumbpipe", "dumbpipe-path") if err != nil { return s, err } s.Dumbpipe = dp.Path return s, nil } func (r *requestRun) proxyCommand() (string, error) { exe, err := os.Executable() if err != nil { return "", fmt.Errorf("resolving this binary's own path for the ProxyCommand: %w", err) } words := append(append([]string{exe}, strings.Fields(cmp.Or(r.verb, "run request"))...), "proxy", r.inv.Name, "--tag", r.tag, "--state-dir", r.stateRoot) if level := strings.TrimSpace(r.vals.String("log-level")); level != "" { words = append(words, "--log-level", level) } if r.target != "" { words = append(words, "--contract", r.target) } for i, w := range words { words[i] = "'" + strings.ReplaceAll(w, "'", `'"'"'`) + "'" } return strings.Join(words, " "), nil } func (r *requestRun) ssh(ctx context.Context, c *v1alpha1.ComputeContract) (int, error) { started := time.Now() s, err := r.sshClient(ctx) if err != nil { return -1, err } t, err := s.Target(c) if err != nil { return -1, err } how := "its iroh ticket" if r.transport == transportWebRTC { how = "a WebRTC datachannel of tunnel " + r.tag + " (" + cmp.Or(r.verb, "run request") + " proxy)" } r.log("ssh to the guest over "+how+", StrictHostKeyChecking=yes against its reported host keys", "alias", t.Alias, "hostKeys", strings.Join(request.SummaryOf(c).HostKeys, ", ")) held, stop := r.whileHeld(ctx) defer stop() if r.transport != transportWebRTC { err = s.WaitReady(held, t, requestSSHReady, func(attempt int, msg string) { r.log("ssh not ready yet", "attempt", attempt, "err", msg) }) } if cause := context.Cause(held); err != nil && cause != nil && ctx.Err() == nil { return -1, cause } if err != nil { return -1, err } interactive := false if r.exec == "" { fi, err := os.Stdin.Stat() if err != nil || fi.Mode()&os.ModeCharDevice == 0 { r.log("no --exec and stdin is not a terminal: the guest answered ssh; no session opened") return 0, nil } interactive = true } code, err := r.runSSH(held, s, t, interactive) r.log("ssh returned", "transport", r.transport, "exit", code, "seconds", time.Since(started).Seconds()) if cause := context.Cause(held); cause != nil && ctx.Err() == nil { r.log("ssh session ended because the contract left Held", "err", cause) return 0, nil } if err != nil { if cause := context.Cause(held); cause != nil && ctx.Err() == nil { r.log("ssh session ended because the contract left Held", "err", cause) return 0, nil } if current, cerr := r.kcp.AwaitContract(ctx, 15*time.Second, leftHeld); cerr == nil { r.log("ssh session ended because the contract left Held", "phase", current.Status.Phase, "err", err) return 0, nil } } if r.hold { if current, cerr := r.kcp.AwaitContract(ctx, 5*time.Second, leftHeld); cerr == nil { r.log("ssh session ended because the contract left Held", "phase", current.Status.Phase) return 0, nil } } r.log("ssh session ended", "exit", code) return code, err } func (r *requestRun) runSSH(ctx context.Context, s request.SSH, t request.Target, interactive bool) (int, error) { deadline := time.Now().Add(requestSSHReady) for attempt := 1; ; attempt++ { began := time.Now() code, err := s.Run(ctx, t, r.exec, interactive, os.Stdin, r.a.stdout, r.a.stderr) if r.transport != transportWebRTC || err != nil || code != sshConnectionFailed || time.Since(began) > requestSSHRetryRun || time.Now().After(deadline) || ctx.Err() != nil { return code, err } r.log("ssh over webrtc did not connect; trying again", "attempt", attempt, "seconds", time.Since(began).Seconds()) select { case <-ctx.Done(): return code, nil case <-time.After(request.SSHPollInterval): } } } func (r *requestRun) whileHeld(ctx context.Context) (context.Context, func()) { held, cancel := context.WithCancelCause(ctx) var wg sync.WaitGroup wg.Go(func() { if c, err := r.kcp.AwaitContract(held, 0, leftHeld); err == nil { cancel(fmt.Errorf("the contract left Held while ssh waited for the guest: phase %s: %s", c.Status.Phase, c.Status.Message)) } }) return held, func() { cancel(nil) wg.Wait() } } var errRevokesPending = errors.New("the requester did not finish revoking the contract's atirc tokens") func releaseSettled(requestedAt string) func(*v1alpha1.ComputeContract) bool { return func(c *v1alpha1.ComputeContract) bool { if c.Status.Phase == v1alpha1.ComputeContractReleased { return true } return c.Status.Phase == v1alpha1.ComputeContractFailed && c.Spec.Release && c.Status.Release != nil && c.Status.Release.RequestedAt == requestedAt } } func revokesPending(c *v1alpha1.ComputeContract) bool { state := c.Status.AtircRevokes return requester.ReleaseRevokesExpected(*c) && state != v1alpha1.AtircRevokesDone && state != v1alpha1.AtircRevokesAbandoned } func (r *requestRun) release(ctx context.Context) (*v1alpha1.ComputeContract, error) { requestedAt := market.FormatTime(time.Now()) if err := r.kcp.Release(ctx, requestedAt); err != nil { return nil, fmt.Errorf("setting spec.release: %w", err) } current, err := r.kcp.AwaitContract(ctx, r.contractTimeout, func(*v1alpha1.ComputeContract) bool { return true }) if err != nil || current == nil { return nil, fmt.Errorf("reading the contract for its release wait within %s: %v", r.contractTimeout, err) } wait := requester.ReleaseWait(*current, r.contractTimeout) r.log("asked the requester to release the contract (spec.release)", "releaseRequestedAt", requestedAt, "timeout", wait.String()) c, err := r.kcp.WaitContract(ctx, wait, releaseSettled(requestedAt), r.progress) if err != nil || c == nil { return c, err } return r.settleRevokes(ctx, c) } func (r *requestRun) releaseForOwner(ctx context.Context) int { r.log("another process runs this invocation and owns its kcp; the release is asked through that kcp and the owner stops it") if err := r.attach(ctx); err != nil { return r.a.fail(exitFailed, "run request release: the invocation's own process holds it but its kcp does not answer: %v", err) } c, err := r.release(ctx) if err != nil || c == nil || c.Status.Phase != v1alpha1.ComputeContractReleased { phase := "" if c != nil { phase = string(c.Status.Phase) } r.a.log.Error("the release did not finish", "err", fmt.Sprint(err), "phase", phase) r.a.log.Error("the owning process keeps the invocation; run `socialweb-computer run request release " + r.inv.Name + "` again to retry the release") return exitFailed } r.log("released; waiting for the owning process to stop the invocation's kcp", "timeout", requestOwnerExitWait.String()) wctx, cancel := context.WithTimeout(ctx, requestOwnerExitWait) defer cancel() if err := request.AwaitOwnerGone(wctx, r.stateRoot, r.inv.Name); err != nil { r.a.log.Warn("the owning process did not exit in time; the contract is Released", "err", err.Error()) } if s, err := request.ReadSummary(r.stateRoot, r.inv.Name); err == nil { r.summary = s } else { r.summary.Phase, r.summary.Message, r.summary.Contract = string(c.Status.Phase), c.Status.Message, request.SummaryOf(c) r.summary.KcpRunning = false } r.a.printJSON(r.summary) return exitOK } func (r *requestRun) followRelease(ctx context.Context) (*v1alpha1.ComputeContract, bool) { current, err := r.kcp.AwaitContract(ctx, requestSummaryRead, func(*v1alpha1.ComputeContract) bool { return true }) if err != nil || current == nil || !current.Spec.Release || releaseSettled(current.Spec.ReleaseRequestedAt)(current) { return current, current != nil } wait := requester.ReleaseWait(*current, r.contractTimeout) r.log("the contract is being released; this process keeps its kcp running until the release settles", "releaseRequestedAt", current.Spec.ReleaseRequestedAt, "timeout", wait.String()) c, err := r.kcp.WaitContract(ctx, wait, releaseSettled(current.Spec.ReleaseRequestedAt), r.progress) if err != nil || c == nil { r.a.log.Error("the release did not settle while this process kept the kcp", "err", fmt.Sprint(err)) return current, true } return c, true } func (r *requestRun) settleRevokes(ctx context.Context, c *v1alpha1.ComputeContract) (*v1alpha1.ComputeContract, error) { if c == nil || !revokesPending(c) { return c, nil } r.log("waiting for the requester to revoke the contract's atirc tokens before its kcp stops", "timeout", cloudinit.RequestRevokeWait.String()) revoked, err := r.kcp.WaitContract(ctx, cloudinit.RequestRevokeWait, func(c *v1alpha1.ComputeContract) bool { return !revokesPending(c) }, r.progress) if err != nil || revoked == nil { return c, fmt.Errorf("%w (atircRevokes %q): %v", errRevokesPending, c.Status.AtircRevokes, err) } r.log("the atirc token revokes finished", "state", string(revoked.Status.AtircRevokes)) return revoked, nil } func (r *requestRun) keptForRevokes(err error) { r.a.log.Error("the invocation's state is kept; run `socialweb-computer run request release "+r.inv.Name+"` again to finish the atirc token revokes", "err", err.Error()) } func (r *requestRun) carryOn(ctx context.Context) int { c, err := r.kcp.WaitContract(ctx, r.contractTimeout, contractSettled, r.progress) if err != nil { r.a.log.Error("the contract did not settle", "err", err.Error()) r.writeSummary(ctx, c) r.stop(context.WithoutCancel(ctx), false) r.a.printJSON(r.summary) return exitFailed } if c.Status.Phase == v1alpha1.ComputeContractHeld && c.Spec.SecretsRef != nil && !secretsApplied(c) { r.log("waiting for the guest to write its secrets (getSecrets, or the sealed fallback when enabled)") if fc, err := r.kcp.WaitContract(ctx, r.contractTimeout, func(c *v1alpha1.ComputeContract) bool { return secretsApplied(c) || c.Status.Phase != v1alpha1.ComputeContractHeld }, r.progress); err != nil { r.a.log.Error("the guest did not write its secrets", "err", err.Error()) } else { c = fc cond := meta.FindStatusCondition(c.Status.Conditions, v1alpha1.ComputeContractConditionSecretsApplied) reason := "" if cond != nil { reason = cond.Reason } r.log("the guest wrote its secrets", "via", reason, "fetchedAt", c.Status.SecretsFetchedAt) } } code := exitOK hasContract := c.Status.Contract != nil && c.Status.Contract.Receipt != nil if c.Status.Phase == v1alpha1.ComputeContractHeld && !r.skipSSH { ec, err := r.ssh(ctx, c) if err != nil { r.a.log.Error("ssh", "err", err.Error()) code = exitFailed } else if ec != 0 { code = exitFailed } } if r.hold && c.Status.Phase == v1alpha1.ComputeContractHeld { if fc, ok := r.followRelease(ctx); ok { c = fc } } if c.Status.Phase == v1alpha1.ComputeContractFailed && !hasContract { code = exitFailed } remove := c.Status.Phase == v1alpha1.ComputeContractReleased || !hasContract if !r.hold && hasContract && c.Status.Phase != v1alpha1.ComputeContractReleased { rc, err := r.release(context.WithoutCancel(ctx)) if rc != nil { c = rc } if err != nil || c.Status.Phase != v1alpha1.ComputeContractReleased { r.a.log.Error("the release did not finish", "err", fmt.Sprint(err), "phase", string(c.Status.Phase)) code = exitFailed } else { remove = true } } if remove { if settled, err := r.settleRevokes(context.WithoutCancel(ctx), c); err != nil { r.keptForRevokes(err) remove, code = false, exitFailed } else { c = settled } } r.writeSummary(ctx, c) r.stop(context.WithoutCancel(ctx), remove) r.writeSummary(ctx, nil) r.a.printJSON(r.summary) return code } func runRequest(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { pos, pass := splitWords(rest) if len(pos) > 0 { return a.fail(exitRefused, "run request: unexpected %q; request-vm-ssh options go after --", strings.Join(pos, " ")) } for _, tool := range []string{"ssh", "ssh-keygen"} { if _, err := exec.LookPath(tool); err != nil { return a.fail(exitRefused, "run request: %s is not on PATH; install the OpenSSH client (apt install openssh-client, dnf install openssh-clients)", tool) } } r, err := newRequestRun(a, vals) if err != nil { return a.fail(exitRefused, "run request: %v", err) } name := vals.String("name") if name == "" { name = "req-" + strings.ToLower(time.Now().UTC().Format("20060102t150405")) } if err := request.CheckName(name); err != nil { return a.fail(exitRefused, "run request: %v", err) } if _, err := os.Stat(request.Dir(r.stateRoot, name)); err == nil { return a.fail(exitRefused, "run request: invocation %s exists under %s; run request resume|release %s", name, request.Root(r.stateRoot), name) } network := vals.String("network") if err := r.passthrough(pass, network != ""); err != nil { return a.fail(exitRefused, "run request: %v", err) } if vals.Bool("guest-requester") { if network != "" { return a.fail(exitRefused, "run request: --guest-requester is for a standalone invocation") } r.guestRequester, r.skipSSH = true, true } if err := r.associationRef(network != ""); err != nil { return a.fail(exitRefused, "run request: --association-ref: %v", err) } if r.contract.ComputeKind == v1alpha1.ComputeKindDeno { r.skipSSH = true } if err := r.webrtcTunnel(); err != nil { return a.fail(exitRefused, "run request: %v", err) } var bundle []byte if r.secrets != "" { if bundle, err = readSecrets(r.secrets); err != nil { return a.fail(exitRefused, "run request: --secrets: %v", err) } } userPaths, err := bundlePaths(bundle) if err != nil { return a.fail(exitRefused, "run request: --secrets: %v", err) } if r.contract.Agent != nil { if bundle, err = withAgentPlaceholder(bundle, r.contract.Agent.EnvFile); err != nil { return a.fail(exitRefused, "run request: --agent: %v", err) } } if r.contract.Association, err = associateOf(cmp.Or(vals.String("associate"), r.associate)); err != nil { return a.fail(exitRefused, "run request: --associate: %v", err) } if r.contract.Association != nil && r.contract.Association.FromGuest && !slices.Contains(r.contract.GuestTools, cloudinit.GuestToolSocialwebComputer) { r.contract.GuestTools = append(r.contract.GuestTools, cloudinit.GuestToolSocialwebComputer) } if r.contract.Association != nil { if bundle, err = withPlaceholder(bundle, cloudinit.DefaultAssociationRefFile, ""); err != nil { return a.fail(exitRefused, "run request: --associate: %v", err) } } if err := requester.CheckTokenPaths(v1alpha1.ComputeContract{Spec: r.contract}, userPaths); err != nil { return a.fail(exitRefused, "run request: %v", err) } r.inv = request.Invocation{Name: name, Mode: request.ModeStandalone, Network: network, StateRoot: r.stateRoot, CreatedAt: time.Now().UTC()} if network != "" { r.inv.Mode = request.ModeNetwork } r.summary = request.Summary{Name: name, Mode: r.inv.Mode, Network: network, CreatedAt: r.inv.CreatedAt} if err := r.openIdentity(name); err != nil { return a.fail(exitRefused, "run request: %v", err) } defer func() { if r.identity != nil { r.identity.Close() } }() if !r.skipSSH { pub, _, err := request.EnsureKey(ctx, filepath.Join(request.Dir(r.stateRoot, name), request.SSHDir)) if err != nil { return a.fail(exitFailed, "run request: %v", err) } r.contract.SshAuthorizedKeys = append([]string{pub}, r.contract.SshAuthorizedKeys...) } if err := request.WriteInvocation(r.inv); err != nil { return a.fail(exitFailed, "run request: %v", err) } owner, owned, err := request.TakeOwner(r.stateRoot, name) if err != nil || !owned { return a.fail(exitFailed, "run request: the invocation %s is owned by another process: %v", name, err) } defer owner.Close() if err := r.linkIdentityExposure(); err != nil { return a.fail(exitFailed, "run request: %v", err) } r.log("a fresh kcp for this invocation", "mode", r.inv.Mode, "network", network) if err := r.start(ctx, true); err != nil { r.writeSummaryQuiet() r.stop(context.WithoutCancel(ctx), false) if ctx.Err() != nil { return a.fail(exitFailed, "run request: stopped before the requester was Ready, so no contract was posted; the invocation's kcp is stopped and its state is kept in %s (delete that directory to drop it)", request.Dir(r.stateRoot, name)) } return a.fail(exitFailed, "run request: %v", err) } if err := r.checkIdentity(); err != nil { r.writeSummaryQuiet() r.stop(context.WithoutCancel(ctx), false) return a.fail(exitFailed, "run request: %v", err) } if bundle != nil { ref, err := r.kcp.PutSecrets(ctx, bundle, r.networkTimeout) if err != nil { r.stop(context.WithoutCancel(ctx), false) return a.fail(exitFailed, "run request: %v", err) } r.contract.SecretsRef = &ref r.log("the secrets bundle is in the invocation's own OpenBao", "mount", ref.Mount, "path", ref.Path) } if err := r.kcp.CreateContract(ctx, r.contract, r.networkTimeout); err != nil { r.stop(context.WithoutCancel(ctx), false) return a.fail(exitFailed, "run request: %v", err) } return r.carryOn(ctx) } func (r *requestRun) checkIdentity() error { if r.identity == nil { return nil } ids, err := r.kcp.Identities() if err != nil { return fmt.Errorf("the identity: %w", err) } ictx, cancel := context.WithTimeout(context.Background(), requestSummaryRead) obj, err := ids.GetMarketIdentity(ictx, marketrequester.RequesterIdentity) cancel() if err != nil { return fmt.Errorf("the identity: %w", err) } r.identityObject = obj expected := r.expectedPDSURL live := strings.TrimSpace(obj.Status.PdsURL) if expected != "" && live != "" && !compose.SamePDSURL(expected, live) { return fmt.Errorf("the identity PDS public URL is %s, not the recorded %s; refusing to post against a changed DID-document endpoint", live, expected) } if err := r.finalizeIdentity(); err != nil { return fmt.Errorf("the identity: %w", err) } r.identity.Close() return nil } func (r *requestRun) writeSummaryQuiet() { if err := request.WriteSummary(r.stateRoot, r.summary); err != nil { r.a.log.Warn("writing summary.json", "err", err.Error()) } } func (a *app) loadRequest(vals clicfg.Values, rest []string, verb string) (*requestRun, int) { pos, pass := splitWords(rest) if len(pos) != 1 || len(pass) > 0 { return nil, a.fail(exitRefused, "run request %s : one invocation name, no request-vm-ssh options", verb) } r, err := newRequestRun(a, vals) if err != nil { return nil, a.fail(exitRefused, "run request %s: %v", verb, err) } inv, err := request.ReadInvocation(r.stateRoot, pos[0]) if err != nil { return nil, a.fail(exitRefused, "run request %s: %v", verb, err) } if inv.StateRoot != "" && inv.StateRoot != r.stateRoot { return nil, a.fail(exitRefused, "run request %s: invocation %s belongs to state root %s, not %s", verb, inv.Name, inv.StateRoot, r.stateRoot) } r.inv = inv if s, err := request.ReadSummary(r.stateRoot, inv.Name); err == nil { r.summary = s } else { r.summary = request.Summary{Name: inv.Name, Mode: inv.Mode, Network: inv.Network, Instance: inv.Instance, CreatedAt: inv.CreatedAt} } if r.summary.Identity == "" { r.summary.Identity = inv.Identity } if inv.Identity != "" { instanceState := request.InstanceStateDir(r.stateRoot, inv.Name) if _, err := os.Stat(instanceState); err != nil { if verb != "release" { return nil, a.fail(exitRefused, "run request %s: invocation %s's kcp state is gone; start a new invocation with --identity %s", verb, inv.Name, inv.Identity) } r.cleanupOnly = true return r, exitOK } if err := request.LinkIdentityExposure(r.stateRoot, inv.Identity, instanceState); err != nil { return nil, a.fail(exitRefused, "run request %s: %v", verb, err) } } return r, exitOK } func (a *app) reopenRequest(ctx context.Context, vals clicfg.Values, rest []string, verb string) (*requestRun, int) { r, code := a.loadRequest(vals, rest, verb) if r == nil { return nil, code } return a.restartLoaded(ctx, r, verb) } func (a *app) restartLoaded(ctx context.Context, r *requestRun, verb string) (*requestRun, int) { if r.cleanupOnly { return r, exitOK } r.log("starting the invocation's kcp again from its state dir", "verb", verb) r.resuming = verb == "resume" if err := r.start(ctx, false); err != nil { r.stop(context.WithoutCancel(ctx), false) return nil, a.fail(exitFailed, "run request %s: %v", verb, err) } return r, exitOK } func runRequestResume(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { r, code := a.loadRequest(vals, rest, "resume") if r == nil { return code } owner, owned, err := request.TakeOwner(r.stateRoot, r.inv.Name) if err != nil { return a.fail(exitFailed, "run request resume: %v", err) } if !owned { return a.fail(exitRefused, "run request resume: another process runs invocation %s; use run request ssh|agent|release %s", r.inv.Name, r.inv.Name) } defer owner.Close() if r, code = a.restartLoaded(ctx, r, "resume"); r == nil { return code } return r.carryOn(ctx) } func runRequestRelease(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { r, code := a.loadRequest(vals, rest, "release") if r == nil { return code } if !r.cleanupOnly { owner, owned, err := request.TakeOwner(r.stateRoot, r.inv.Name) if err != nil { return a.fail(exitFailed, "run request release: %v", err) } if !owned { return r.releaseForOwner(ctx) } defer owner.Close() } if r, code = a.restartLoaded(ctx, r, "release"); r == nil { return code } if r.cleanupOnly { r.stop(context.WithoutCancel(ctx), true) if err := os.RemoveAll(request.Dir(r.stateRoot, r.inv.Name)); err != nil { return a.fail(exitFailed, "run request release: removing the invocation: %v", err) } r.log("the invocation's kcp state was already gone; the invocation is removed and the identity store is kept") r.a.printJSON(r.summary) return exitOK } c, err := r.release(ctx) if err != nil || c == nil || c.Status.Phase != v1alpha1.ComputeContractReleased { phase := "" if c != nil { phase = string(c.Status.Phase) } r.a.log.Error("the release did not finish", "err", fmt.Sprint(err), "phase", phase) if errors.Is(err, errRevokesPending) { r.keptForRevokes(err) } else { r.a.log.Error("the invocation's state is kept; run `socialweb-computer run request release " + r.inv.Name + "` again to retry the release") } r.writeSummary(ctx, c) r.stop(context.WithoutCancel(ctx), false) r.writeSummary(ctx, nil) r.a.printJSON(r.summary) return exitFailed } r.writeSummary(ctx, c) r.stop(context.WithoutCancel(ctx), true) if r.standalone.StateDir != "" { if _, err := os.Stat(r.standalone.StateDir); err == nil { r.writeSummary(ctx, nil) r.a.printJSON(r.summary) return a.fail(exitFailed, "run request release: the invocation's processes remain; its state is kept") } } r.writeSummary(ctx, nil) r.a.printJSON(r.summary) return exitOK } func runRequestList(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { root, err := devinstance.StateRoot(vals.String("state-dir"), a.env) if err != nil { return a.fail(exitRefused, "run request list: %v", err) } if root, err = filepath.Abs(root); err != nil { return a.fail(exitRefused, "run request list: %v", err) } list, err := request.List(root) if err != nil { return a.fail(exitFailed, "run request list: %v", err) } if vals.Bool("json") { a.printJSON(list) return exitOK } for _, s := range list { c := request.ContractSummary{} if s.Contract != nil { c = *s.Contract } where := s.Mode if s.Network != "" { where = "network " + s.Network } extra := "" if c.R2Release != "" { extra += " r2Release=" + strconv.Quote(c.R2Release) } if c.AtircRevokes != "" { extra += " atircRevokes=" + c.AtircRevokes } fmt.Fprintf(a.stdout, "%-24s %-20s %-10s kcp=%-5t identity=%s did=%s winner=%s guest=%s %s%s\n", s.Name, where, cmp.Or(s.Phase), s.KcpRunning, cmp.Or(s.Identity, "-"), cmp.Or(s.Did, "-"), cmp.Or(c.WinnerDid), cmp.Or(c.VMAddress), s.Message, extra) } if len(list) == 0 { fmt.Fprintln(a.stdout, "no invocations under", request.Root(root)) } return exitOK } func (r *requestRun) keepLogs() { if r.instanceDir == "" { return } b, err := os.ReadFile(filepath.Join(r.instanceDir, "logs", "core.log")) if err != nil { return } out := filepath.Join(r.home(), "logs") if err := os.MkdirAll(out, 0o700); err != nil { return } os.WriteFile(filepath.Join(out, "core.log"), b, 0o600) } func (r *requestRun) attach(ctx context.Context) error { ctx, cancel := context.WithTimeout(ctx, requestAttachProbe) defer cancel() var kubeconfig, workspace string switch r.inv.Mode { case request.ModeNetwork: netLayout, netRec, err := r.networkLayout() if err != nil { return err } app, ok, err := request.NetworkApp(ctx, netRec.AdminKubeconfig, r.inv.Name) if err != nil { return err } if !ok || app.Instance == "" || app.Message == marketnetwork.StoppedMessage { return errors.New("the market network does not run the invocation") } rec, err := r.childRecord(netLayout, app.Instance) if err != nil { return err } r.instanceDir = rec.Layout.StateDir kubeconfig, workspace = rec.AdminKubeconfig, rec.BaseWorkspace default: l, err := r.a.layoutOf(r.inv.Name, r.vals.With("instance", r.inv.Name).With("state-dir", r.home())) if err != nil { return err } var rec devinstance.Record if err := fsx.ReadJSON(l.Record, &rec); err != nil { return err } if rec.AdminKubeconfig == "" { return errors.New("the invocation's kcp has no admin kubeconfig") } r.standalone, r.instanceDir = l, l.StateDir kubeconfig = rec.AdminKubeconfig } k, err := request.OpenKcp(kubeconfig, workspace) if err != nil { return err } st, err := k.Root(ctx) if err != nil { return err } if st.Phase != v1alpha1.ApplicationPhaseReady { return fmt.Errorf("the MarketRequester is %s", st.Phase) } r.kcp = k.ForContract(r.target) return nil } func (a *app) openRequest(ctx context.Context, vals clicfg.Values, rest []string, verb string) (*requestRun, bool, int) { r, code := a.loadRequest(vals, rest, verb) if r == nil { return nil, false, code } err := r.attach(ctx) if err == nil { return r, false, exitOK } r.log("the invocation's kcp does not answer; starting it again from its state dir", "verb", verb, "why", err.Error()) if err := r.start(ctx, false); err != nil { r.stop(context.WithoutCancel(ctx), false) return nil, false, a.fail(exitFailed, "run request %s: %v", verb, err) } return r, true, exitOK } func runRequestSSH(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { r, started, code := a.openRequest(ctx, vals, rest, "ssh") if r == nil { return code } if started { defer r.stop(context.WithoutCancel(ctx), false) } c, err := r.kcp.WaitContract(ctx, r.contractTimeout, contractSettled, r.progress) if err != nil { return a.fail(exitFailed, "run request ssh: %v", err) } if c.Status.Phase != v1alpha1.ComputeContractHeld { return a.fail(exitFailed, "run request ssh: the contract is %s, not Held: %s", c.Status.Phase, c.Status.Message) } ec, err := r.ssh(ctx, c) if err != nil { return a.fail(exitFailed, "run request ssh: %v", err) } if ec != 0 { return exitFailed } return exitOK } func runRequestProxy(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { return proxyRequest(ctx, a, vals, "run request proxy", func(q *app) (*requestRun, bool, int) { return q.openRequest(ctx, vals, rest, "proxy") }) } func proxyRequest(ctx context.Context, a *app, vals clicfg.Values, verb string, open func(*app) (*requestRun, bool, int)) int { level, err := logx.ParseLevel(vals.String("log-level")) if err != nil { return a.fail(exitRefused, "%s: --log-level: %v", verb, err) } quiet := *a quiet.stdout = a.stderr handler := &slog.HandlerOptions{Level: level} if level >= slog.LevelWarn { handler.ReplaceAttr = dropTime } quiet.log = slog.New(slog.NewTextHandler(a.stderr, handler)).With("service", serviceName) r, started, code := open(&quiet) if r == nil { return code } if started { defer r.stop(context.WithoutCancel(ctx), false) } if r.tag == "" { return a.fail(exitRefused, "%s: --tag names the webrtc tunnel", verb) } log := quiet.log.With("request", r.inv.Name) conn, timings, err := webrtcproxy.Dial(ctx, r.kcp.Signaling(), webrtcproxy.Options{Tag: r.tag, Log: log}) if err != nil { return a.fail(exitFailed, "%s: %s", verb, strings.Join(strings.Fields(err.Error()), " ")) } defer conn.Close() log.Info("the proxy pipes stdin and stdout through the datachannel", "tag", r.tag, "guestSeconds", timings.Guest.Seconds(), "offerToAnswerSeconds", timings.Answer.Seconds(), "offerToOpenSeconds", timings.Open.Seconds()) if err := webrtcproxy.Pipe(ctx, conn, os.Stdin, a.stdout); err != nil && ctx.Err() == nil { log.Warn("the proxy ends", "err", err.Error()) } return exitOK } func dropTime(groups []string, attr slog.Attr) slog.Attr { if len(groups) == 0 && attr.Key == slog.TimeKey { return slog.Attr{} } return attr }