package main import ( "cmp" "context" "fmt" "net" "slices" "strconv" "strings" "time" "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/sshgateway" "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/hostsvc" "github.com/johnandersen777/socialweb-computer-kcp/internal/irohdial" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx" "github.com/johnandersen777/socialweb-computer-kcp/internal/stack/gatewayssh" "github.com/johnandersen777/socialweb-computer-kcp/internal/stack/request" ) var gatewaySshOptions = slices.Concat([]string{backendOpt, "profile", "network", "name", "instance", "state-dir", "fg", "ready-timeout", "network-timeout", "kcp-path", "kine-path", "bao-path", "dumbpipe-path", "asset-cache"}) type gatewaySshRun struct { a *app vals clicfg.Values profile string network string options []v1alpha1.ApplicationOption comms *v1alpha1.MarketComms value *v1alpha1.MarketNetworkValue bidderDids []string ownerSessions string requesterIngress *v1alpha1.IngressConfig ssh *v1alpha1.SshGatewaySsh networkTimeout time.Duration } type gatewaySshSummary struct { Profile string `json:"profile"` Network string `json:"network,omitempty"` Name string `json:"name,omitempty"` Instance string `json:"instance,omitempty"` KcpURL string `json:"kcpUrl,omitempty"` Phase v1alpha1.ApplicationPhase `json:"phase,omitempty"` Message string `json:"message,omitempty"` SSH string `json:"ssh,omitempty"` Listeners []v1alpha1.ApplicationListener `json:"listeners,omitempty"` Components []v1alpha1.ApplicationComponent `json:"components,omitempty"` Seconds float64 `json:"seconds"` } func (r *gatewaySshRun) log(msg string, args ...any) { r.a.log.Info(msg, append([]any{"verb", "run gateway ssh", "profile", r.profile, "utc", time.Now().UTC().Format(time.RFC3339)}, args...)...) } func (r *gatewaySshRun) passthrough(args []string) error { parsed, err := compose.ParseAppArgs(sshgateway.SshGatewayApp, args) if err != nil { return err } var nv v1alpha1.MarketNetworkValue nvSet := false for _, p := range parsed { switch p.Name { case "test-owner-sessions-file": r.ownerSessions = p.Value continue case "bidder-dids": r.bidderDids = append(r.bidderDids, splitList(p.Value)...) continue case "ssh-public-listen", "ssh-web-listen", "ssh-operator-did", "ssh-max-accounts": if err := r.applySshOption(p.Name, p.Value); err != nil { return err } continue case "plc-directory-url", "ingress-proxy-host", "network-hosts": if r.network != "" { return fmt.Errorf("--%s is set by the market network for its child applications", p.Name) } nvSet = true switch p.Name { case "plc-directory-url": nv.PLCDirectoryURL = p.Value case "ingress-proxy-host": nv.IngressProxyHost = p.Value case "network-hosts": for _, h := range splitList(p.Value) { name, addr, ok := strings.Cut(h, "=") if !ok { return fmt.Errorf("--network-hosts %q: want name=address,...", p.Value) } nv.Hosts = append(nv.Hosts, v1alpha1.NetworkHost{Name: name, Address: addr}) } } continue } if p.To == compose.MapToOptions { r.options = append(r.options, v1alpha1.ApplicationOption{Name: p.Name, Value: p.Value}) continue } if strings.HasPrefix(p.To, "spec.comms") { var applied bool var err error if r.comms, applied, err = applyCommsOption(r.comms, p.Name, p.Value); err != nil { return err } if !applied { return fmt.Errorf("--%s maps to %s, which run gateway ssh does not take", p.Name, p.To) } continue } if strings.HasPrefix(p.To, "spec.requesterIngress") { var applied bool var err error if r.requesterIngress, applied, err = applyIngressOption(r.requesterIngress, p.Name, p.Value); err != nil { return err } if !applied { return fmt.Errorf("--%s maps to %s, which run gateway ssh does not take", p.Name, p.To) } continue } return fmt.Errorf("--%s maps to %s, which run gateway ssh does not take", p.Name, p.To) } if nvSet { r.value = &nv } return nil } func (r *gatewaySshRun) applySshOption(name, value string) error { if r.ssh == nil { r.ssh = &v1alpha1.SshGatewaySsh{} } value = strings.TrimSpace(value) switch name { case "ssh-operator-did": if !strings.HasPrefix(value, "did:") || strings.ContainsAny(value, " /") { return fmt.Errorf("--ssh-operator-did %q is not a DID", value) } r.ssh.OperatorDid = value case "ssh-max-accounts": n, err := strconv.Atoi(value) if err != nil || n < 1 { return fmt.Errorf("--ssh-max-accounts %q: want a positive number", value) } r.ssh.MaxAccounts = int32(n) case "ssh-public-listen", "ssh-web-listen": host, port, err := net.SplitHostPort(value) if err != nil { return fmt.Errorf("--%s %q: want host:port: %w", name, value, err) } if n, err := strconv.Atoi(port); err != nil || n < 1 || n > 65535 { return fmt.Errorf("--%s %q: want a port from 1 to 65535", name, value) } if name == "ssh-web-listen" { r.ssh.WebListen = net.JoinHostPort(host, port) } else { r.ssh.PublicListen = net.JoinHostPort(host, port) } } return nil } func runGatewaySsh(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { pos, pass := splitWords(rest) if len(pos) > 0 { return a.fail(exitRefused, "run gateway ssh: unexpected %q; hono-socialweb-computer-ssh options go after --", strings.Join(pos, " ")) } r := &gatewaySshRun{a: a, vals: vals, profile: vals.String("profile"), network: vals.String("network")} var err error if r.networkTimeout, err = time.ParseDuration(vals.String("network-timeout")); err != nil { return a.fail(exitRefused, "run gateway ssh: network-timeout: %v", err) } if r.network != "" && (r.profile == "" || r.profile == sshgateway.SshGatewayProfileDefault) { r.profile = sshgateway.SshGatewayProfileNetwork } r.profile = cmp.Or(r.profile, sshgateway.SshGatewayProfileDefault) if !slices.Contains(sshgateway.SshGatewayProfiles, r.profile) { return a.fail(exitRefused, "run gateway ssh: --profile %q: want one of %s", r.profile, strings.Join(sshgateway.SshGatewayProfiles, ", ")) } if err := r.passthrough(pass); err != nil { return a.fail(exitRefused, "run gateway ssh: %v", err) } backend, err := hostsvc.ParseBackend(vals.String(backendOpt)) if err != nil { return a.fail(exitRefused, "run gateway ssh: %v", err) } switch { case r.network != "" && backend != hostsvc.BackendProc: return a.fail(exitRefused, "run gateway ssh: --network: the market network places its children under its own devsup; pass --host-backend=proc") case r.network != "" && r.ssh != nil: return a.fail(exitRefused, "run gateway ssh: the --ssh-* options are options of a standalone ssh gateway; a market network child listens on loopback only") case r.network != "": return r.inNetwork(ctx) case r.profile == sshgateway.SshGatewayProfileNetwork: return a.fail(exitRefused, "run gateway ssh: --profile %s needs --network (dev up --profile market-network)", r.profile) } return r.standalone(ctx) } func (r *gatewaySshRun) name() (string, error) { name := r.vals.String("name") name = cmp.Or(name, gatewayssh.DefaultName) return name, request.CheckName(name) } func (r *gatewaySshRun) report(sum gatewaySshSummary, st v1alpha1.ApplicationStatus) gatewaySshSummary { sum.Phase, sum.Message, sum.Listeners, sum.Components = st.Phase, st.Message, st.Listeners, st.Components sum.SSH = gatewayssh.SshListener(st.Listeners) return sum } func (r *gatewaySshRun) inNetwork(ctx context.Context) int { started := time.Now() l, err := r.a.layoutOf(r.network, r.vals) if err != nil { return r.a.fail(exitRefused, "run gateway ssh: %v", err) } var netRec devinstance.Record if err := fsx.ReadJSON(l.Record, &netRec); err != nil { return r.a.fail(exitRefused, "run gateway ssh: market network %s has no instance record (dev up --profile market-network first): %v", r.network, err) } if netRec.Profile != bootstrap.ProfileMarketNetwork || netRec.AdminKubeconfig == "" { return r.a.fail(exitRefused, "run gateway ssh: instance %s is not a market network (profile %s)", r.network, netRec.Profile) } if netRec.Executables[sshgateway.DumbpipeExecutable].Path == "" { return r.a.fail(exitRefused, "run gateway ssh: market network %s was brought up without dumbpipe; its children reach guests over iroh only (dev up --dumbpipe-path in a dev build)", r.network) } name, err := r.name() if err != nil { return r.a.fail(exitRefused, "run gateway ssh: --name: %v", err) } app := v1alpha1.NetworkApp{Name: name, Kind: v1alpha1.ApplicationKindSshGateway, Profile: r.profile, Options: r.options, OwnerSessionsFile: r.ownerSessions, RequesterIngress: r.requesterIngress} cur, found, err := request.NetworkApp(ctx, netRec.AdminKubeconfig, name) switch { case err != nil: return r.a.fail(exitFailed, "run gateway ssh: %v", err) case !found: if err := request.AddToNetwork(ctx, netRec.AdminKubeconfig, app); err != nil { return r.a.fail(exitFailed, "run gateway ssh: %v", err) } r.log("the market network places the ssh gateway in its workspace of the network's kcp", "network", r.network, "name", name) case cur.Message == marketnetwork.StoppedMessage: if err := request.SetStopped(ctx, netRec.AdminKubeconfig, name, false); err != nil { return r.a.fail(exitFailed, "run gateway ssh: %v", err) } r.log("the market network starts the ssh gateway's kcp again from its kept state", "name", name) default: r.log("the market network already runs this ssh gateway; waiting for it to be Ready", "name", name, "phase", string(cur.Phase)) } st, err := request.WaitNetworkApp(ctx, netRec.AdminKubeconfig, name, r.networkTimeout, func(a v1alpha1.NetworkAppStatus) bool { return a.Phase == v1alpha1.ApplicationPhaseReady && gatewayssh.SshListener(a.Listeners) != "" }, func(a v1alpha1.NetworkAppStatus) { r.log("network child", "name", name, "phase", string(a.Phase), "message", a.Message) }) sum := r.report(gatewaySshSummary{Profile: r.profile, Network: r.network, Name: name, Instance: st.Instance, KcpURL: st.KcpURL}, v1alpha1.ApplicationStatus{Phase: st.Phase, Message: st.Message, Listeners: st.Listeners, Components: st.Components}) sum.Seconds = time.Since(started).Seconds() r.a.printJSON(sum) if err != nil { return r.a.fail(exitFailed, "run gateway ssh: %v", err) } if !r.vals.Bool("fg") { return exitOK } r.log("run gateway ssh --fg: the ssh gateway lives as long as this process; Ctrl-C removes it from the market network", "name", name) <-ctx.Done() rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), networkDownTimeout) defer cancel() if err := request.RemoveFromNetwork(rctx, netRec.AdminKubeconfig, name); err != nil { return r.a.fail(exitFailed, "run gateway ssh: removing %s from the market network: %v", name, err) } return exitOK } func (r *gatewaySshRun) standalone(ctx context.Context) int { started := time.Now() name, err := r.name() if err != nil { return r.a.fail(exitRefused, "run gateway ssh: --name: %v", err) } spec, err := gatewayssh.Spec(gatewayssh.SpecInput{Profile: r.profile, Network: r.value, BidderDids: r.bidderDids, Options: r.options, Comms: r.comms, RequesterIngress: r.requesterIngress, Ssh: r.ssh}) if err != nil { return r.a.fail(exitRefused, "run gateway ssh: %v", err) } instance := r.vals.String("instance") if instance == "" { instance = gatewayssh.InstanceID(name) } vals := r.vals.With("instance", instance).With("profile", bootstrap.ProfileQuick) if l, err := r.a.layoutOf(instance, vals); err != nil { return r.a.fail(exitRefused, "run gateway ssh: %v", err) } else if err := sshgateway.CheckIrohSocket(l); err != nil { return r.a.fail(exitRefused, "run gateway ssh: %v", err) } up, code := bringUp(ctx, r.a, vals, false) if code != exitOK { return code } if up.Executables[sshgateway.DumbpipeExecutable].Path == "" { return r.a.fail(exitRefused, "run gateway ssh: the instance has no dumbpipe; the ssh gateway reaches guests over iroh only (--dumbpipe-path in a dev build)") } k, err := gatewayssh.OpenKcp(up.AdminKubeconfig) if err != nil { return r.a.fail(exitFailed, "run gateway ssh: %v", err) } if err := k.ApplyRoot(ctx, spec); err != nil { return r.a.fail(exitFailed, "run gateway ssh: applying the SshGateway root object: %v", err) } r.log("the SshGateway root object is applied into its own kcp", "instance", up.Instance, "kcp", up.KcpURL) o, err := k.WaitReady(ctx, r.networkTimeout, func(st v1alpha1.ApplicationStatus) { r.log("ssh gateway", "phase", string(st.Phase), "message", st.Message) }) sum := gatewaySshSummary{Profile: r.profile, Name: name, Instance: up.Instance, KcpURL: up.KcpURL} if o != nil { sum = r.report(sum, o.Status) } sum.Seconds = time.Since(started).Seconds() r.a.printJSON(sum) if err != nil { return r.a.fail(exitFailed, "run gateway ssh: %v", err) } if !r.vals.Bool("fg") { return exitOK } r.log("run gateway ssh --fg: the ssh gateway lives as long as this process; Ctrl-C stops its instance and keeps its state", "instance", up.Instance) <-ctx.Done() dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), fgDownTimeout) defer cancel() d := downInstance(dctx, r.a, up.layout, true, false) r.a.printJSON(d) if len(d.Remaining) > 0 { return exitFailed } return exitOK } func runIrohDial(ctx context.Context, a *app, vals clicfg.Values, rest []string) int { if len(rest) > 0 { return a.fail(exitRefused, "internal iroh-dial: unexpected %q", strings.Join(rest, " ")) } err := irohdial.Run(ctx, irohdial.Config{Socket: vals.String("socket"), Dumbpipe: vals.String("dumbpipe-path"), Log: a.log}) if err != nil { return a.fail(exitFailed, "internal iroh-dial: %v", err) } return exitOK }