package opentunnel import ( "bufio" "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "io" "log/slog" "net" "net/url" "os" "os/exec" "path/filepath" "regexp" "strings" "sync" "time" "github.com/johnandersen777/socialweb-computer-kcp/internal/assets" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/component" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/ingress" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/childproc" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/clock" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/loopback" ) const ( DefaultListenHost = "127.0.0.1" DefaultCommandPath = "/usr/local/bin/opentunnel" URLTimeout = 30 * time.Second MinRestartBackoff = time.Second MaxRestartBackoff = 30 * time.Second ) var ( urlPattern = regexp.MustCompile(`https://[^\s"'<>]+`) routePattern = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?$`) ) type Config struct { Command []string RouteName string ProxyProtocol string LocalHost string StateDir string Log *slog.Logger } var ( _ ingress.Provider = (*Provider)(nil) _ ingress.Registrar = (*Provider)(nil) _ ingress.Registration = (*registration)(nil) ) func (c Config) Check() error { command := c.command() path := strings.TrimSpace(command[0]) switch { case path == "": return errors.New("ingressproxy opentunnel: spec.opentunnel.command[0] is empty") case filepath.IsAbs(path): st, err := os.Stat(path) if err != nil { return fmt.Errorf("ingressproxy opentunnel: command: %w", err) } if st.IsDir() || st.Mode().Perm()&0o111 == 0 { return fmt.Errorf("ingressproxy opentunnel: command %s is not an executable file", path) } default: if _, err := exec.LookPath(path); err != nil { return fmt.Errorf("ingressproxy opentunnel: command %q: %w", path, err) } } host := strings.TrimSpace(c.LocalHost) if host != "" { if strings.ContainsAny(host, ":/[]/ \t") { return fmt.Errorf("ingressproxy opentunnel: localHost %q is a host without a port", host) } if lower := strings.ToLower(host); net.ParseIP(host) == nil && (len(lower) > 63 || !routePattern.MatchString(lower)) { return fmt.Errorf("ingressproxy opentunnel: localHost %q is not a host name or IP address", host) } } return nil } func (c Config) command() []string { if len(c.Command) > 0 { return append([]string(nil), c.Command...) } if path := strings.TrimSpace(os.Getenv("SOCIALWEB_OPENTUNNEL_PATH")); path != "" { return []string{path} } if st, err := os.Stat(DefaultCommandPath); err == nil && !st.IsDir() && st.Mode().Perm()&0o111 != 0 { return []string{DefaultCommandPath} } return []string{"opentunnel"} } func (p *Provider) usePinnedCommand(log *slog.Logger) error { if len(p.cfg.Command) > 0 || strings.TrimSpace(os.Getenv("SOCIALWEB_OPENTUNNEL_PATH")) != "" { return nil } if st, err := os.Stat(DefaultCommandPath); err == nil && !st.IsDir() && st.Mode().Perm()&0o111 != 0 { return nil } cache := "" if dir := strings.TrimSpace(p.cfg.StateDir); dir != "" { cache = filepath.Join(dir, "runtime") } executable, err := assets.Resolve(assets.Options{Cache: cache, Log: log}, "opentunnel") if err != nil { return fmt.Errorf("ingressproxy opentunnel: resolving the pinned asset: %w", err) } p.cfg.Command = []string{executable.Path} return nil } func routeName(configured string, spec ingress.Spec) string { if name := strings.TrimSpace(configured); name != "" { return name } label := strings.ToLower(spec.Workspace + "-" + spec.Exposure) var b strings.Builder lastDash := false for _, r := range label { switch { case r >= 'a' && r <= 'z', r >= '0' && r <= '9': b.WriteRune(r) lastDash = false default: if !lastDash && b.Len() > 0 { b.WriteByte('-') lastDash = true } } } name := strings.Trim(b.String(), "-") if len(name) > 63 { name = strings.TrimRight(name[:63], "-") } if name == "" { return "pds" } return name } func (c Config) localHost() string { host := strings.TrimSpace(c.LocalHost) if host == "" { return DefaultListenHost } return host } type Provider struct { cfg Config mu sync.Mutex log *slog.Logger args []string ctx context.Context cancel context.CancelFunc stopCh chan struct{} stopOnce sync.Once ready bool readiness component.ReadySignal registrations map[*registration]struct{} wg sync.WaitGroup } func New(cfg Config) *Provider { return &Provider{cfg: cfg} } func (p *Provider) Start(ctx context.Context) error { log := p.cfg.Log if log == nil { log = slog.New(slog.DiscardHandler) } if err := p.usePinnedCommand(log); err != nil { return err } if err := p.cfg.Check(); err != nil { return err } args := p.cfg.command() path := args[0] if !filepath.IsAbs(path) { resolved, err := exec.LookPath(path) if err != nil { return fmt.Errorf("ingressproxy opentunnel: command %q: %w", path, err) } path = resolved } args[0] = path runCtx, cancel := context.WithCancel(ctx) p.mu.Lock() if p.ctx != nil { p.mu.Unlock() cancel() return errors.New("ingressproxy opentunnel: the provider already started") } p.log = log p.args = args p.ctx = runCtx p.cancel = cancel p.stopCh = make(chan struct{}) p.stopOnce = sync.Once{} p.ready = true p.registrations = map[*registration]struct{}{} stopCh := p.stopCh p.mu.Unlock() log.Info("ingressproxy_opentunnel_started", "provider", "opentunnel", "command", strings.Join(args, " "), "localHost", p.cfg.localHost(), "route", p.cfg.RouteName, "proxyProtocol", p.cfg.ProxyProtocol) p.readiness.Signal() select { case <-ctx.Done(): case <-stopCh: } cancel() p.mu.Lock() p.ready = false p.ctx = nil p.cancel = nil p.mu.Unlock() p.closeAll() p.wg.Wait() return nil } func (p *Provider) Ready() component.Readiness { return &p.readiness } func (p *Provider) Stop(ctx context.Context) error { p.mu.Lock() stopCh := p.stopCh p.mu.Unlock() if stopCh == nil { return nil } p.stopOnce.Do(func() { close(stopCh) }) p.closeAll() return nil } func (p *Provider) Endpoints() []component.Endpoint { return nil } func (p *Provider) Register(ctx context.Context, spec ingress.Spec, target ingress.Target) (ingress.Registration, error) { p.mu.Lock() ready := p.ready root := p.ctx args := append([]string(nil), p.args...) log := p.log p.mu.Unlock() if !ready || root == nil || len(args) == 0 { return nil, errors.New("ingressproxy opentunnel: the provider is not started") } if target == nil { return nil, errors.New("ingressproxy opentunnel: the handler registry is not configured") } bridge, err := loopback.Start(root, loopback.Config{ Host: p.cfg.localHost(), Protocol: spec.Protocol, }, target) if err != nil { return nil, err } runCtx, cancel := context.WithCancel(root) reg := ®istration{ provider: p, spec: spec, bridge: bridge, args: args, log: log.With("workspace", spec.Workspace, "exposure", spec.Exposure), cancel: cancel, done: make(chan struct{}), } child, err := reg.startChild(runCtx) if err != nil { cancel() _ = bridge.Close() return nil, err } if err := p.rememberURL(spec, child.url); err != nil { child.stop() cancel() _ = bridge.Close() return nil, err } reg.mu.Lock() reg.cmd = child.cmd reg.url = child.url + spec.Path reg.mu.Unlock() p.mu.Lock() running := p.ctx == root if running { if p.registrations == nil { p.registrations = map[*registration]struct{}{} } p.registrations[reg] = struct{}{} p.wg.Go(func() { reg.supervise(runCtx, child) }) } p.mu.Unlock() if !running { cancel() child.stop() _ = bridge.Close() return nil, errors.New("ingressproxy opentunnel: the provider stopped") } return reg, nil } func (p *Provider) closeAll() { p.mu.Lock() registrations := p.registrations p.registrations = nil p.mu.Unlock() for registration := range registrations { _ = registration.Close() } } func (p *Provider) forget(registration *registration) { p.mu.Lock() if p.registrations != nil { delete(p.registrations, registration) } p.mu.Unlock() } func (p *Provider) rememberURL(spec ingress.Spec, raw string) error { origin := originOf(raw) if origin == "" { return fmt.Errorf("ingressproxy opentunnel: printed URL %q has no https origin", raw) } dir := strings.TrimSpace(p.cfg.StateDir) if dir == "" { return nil } sum := sha256.Sum256([]byte(spec.Workspace + "/" + spec.Exposure)) path := filepath.Join(dir, "opentunnel", hex.EncodeToString(sum[:8])+".url") current, err := os.ReadFile(path) switch { case err == nil: if previous := strings.TrimSpace(string(current)); previous != "" && !strings.EqualFold(previous, origin) { return fmt.Errorf("%w: opentunnel route %q for %s/%s is %s, was %s", ingress.ErrStableHostMismatch, routeName(p.cfg.RouteName, spec), spec.Workspace, spec.Exposure, origin, previous) } return nil case !errors.Is(err, os.ErrNotExist): return fmt.Errorf("ingressproxy opentunnel: reading the stable host %s: %w", path, err) } if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { return fmt.Errorf("ingressproxy opentunnel: the stable host dir: %w", err) } tmp := path + ".tmp" if err := os.WriteFile(tmp, []byte(origin+"\n"), 0o600); err != nil { return fmt.Errorf("ingressproxy opentunnel: writing the stable host: %w", err) } if err := os.Rename(tmp, path); err != nil { _ = os.Remove(tmp) return fmt.Errorf("ingressproxy opentunnel: placing the stable host: %w", err) } return nil } func (p *Provider) rememberRestart(spec ingress.Spec, raw string) error { return p.rememberURL(spec, raw) } func originOf(raw string) string { parsed, err := url.Parse(strings.TrimSpace(raw)) if err != nil || parsed.Scheme != "https" || parsed.Host == "" { return "" } return parsed.Scheme + "://" + parsed.Host } type registration struct { provider *Provider spec ingress.Spec bridge *loopback.Bridge args []string log *slog.Logger cancel context.CancelFunc done chan struct{} closeOnce sync.Once mu sync.Mutex cmd *exec.Cmd url string } func (r *registration) URL() string { r.mu.Lock() defer r.mu.Unlock() return r.url } func (r *registration) Provider() string { return "opentunnel (https URL)" } func (r *registration) Close() error { r.closeOnce.Do(func() { r.cancel() r.mu.Lock() cmd := r.cmd r.mu.Unlock() if cmd != nil && cmd.Process != nil { _ = cmd.Process.Kill() } _ = r.bridge.Close() <-r.done r.provider.forget(r) }) return nil } type child struct { cmd *exec.Cmd url string tail *childproc.Tail started time.Time output *sync.WaitGroup } func (c *child) stop() { childproc.Kill(c.cmd) c.output.Wait() } func (r *registration) startChild(ctx context.Context) (*child, error) { args := append([]string(nil), r.args...) if _, _, err := net.SplitHostPort(r.bridge.Address()); err != nil { return nil, fmt.Errorf("ingressproxy opentunnel: the loopback address %q: %w", r.bridge.Address(), err) } args = append(args, "route", "add", r.bridge.Address()) args = append(args, "--name", routeName(r.provider.cfg.RouteName, r.spec)) switch strings.ToLower(strings.TrimSpace(r.provider.cfg.ProxyProtocol)) { case "v1", "v2": args = append(args, "--proxy-protocol", strings.ToLower(strings.TrimSpace(r.provider.cfg.ProxyProtocol))) } cmd := exec.CommandContext(ctx, args[0], args[1:]...) cmd.Env = os.Environ() stdout, err := cmd.StdoutPipe() if err != nil { return nil, fmt.Errorf("ingressproxy opentunnel: stdout: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { return nil, fmt.Errorf("ingressproxy opentunnel: stderr: %w", err) } if err := cmd.Start(); err != nil { return nil, fmt.Errorf("ingressproxy opentunnel: starting %s: %w", args[0], err) } tail := &childproc.Tail{} found := make(chan string, 1) var pumps sync.WaitGroup pumps.Go(func() { pump(stdout, tail, found) }) pumps.Go(func() { pump(stderr, tail, found) }) output := &sync.WaitGroup{} output.Go(func() { pumps.Wait() close(found) }) stop := func() { childproc.Kill(cmd) output.Wait() } timer := time.NewTimer(URLTimeout) defer timer.Stop() for { select { case <-ctx.Done(): stop() return nil, ctx.Err() case <-timer.C: stop() return nil, fmt.Errorf("ingressproxy opentunnel: %s printed no URL within %s: %s", args[0], URLTimeout, strings.TrimSpace(tail.String())) case raw, ok := <-found: if !ok { stop() return nil, fmt.Errorf("ingressproxy opentunnel: %s ended before printing a URL: %s", args[0], strings.TrimSpace(tail.String())) } r.log.Info("ingressproxy_opentunnel_route_added", "address", r.bridge.Address(), "url", raw) return &child{cmd: cmd, url: raw, tail: tail, started: time.Now(), output: output}, nil } } } func (r *registration) supervise(ctx context.Context, first *child) { defer close(r.done) current := first backoff := MinRestartBackoff for { if current == nil { next, err := r.startChild(ctx) if err != nil { if ctx.Err() != nil { return } r.log.Warn("ingressproxy_opentunnel_restart_failed", "err", err.Error()) backoff = childproc.NextBackoff(backoff, MinRestartBackoff, MaxRestartBackoff) if !clock.Sleep(ctx, backoff) { return } continue } if err := r.provider.rememberRestart(r.spec, next.url); err != nil { next.stop() r.log.Error("ingressproxy_opentunnel_url_changed", "workspace", r.spec.Workspace, "exposure", r.spec.Exposure, "err", err.Error()) return } current = next r.mu.Lock() r.cmd = next.cmd r.url = next.url + r.spec.Path r.mu.Unlock() } started := current.started err := current.cmd.Wait() current.output.Wait() stderr := strings.TrimSpace(current.tail.String()) r.mu.Lock() if r.cmd == current.cmd { r.cmd = nil } r.mu.Unlock() current = nil if ctx.Err() != nil { return } if time.Since(started) >= time.Minute { backoff = MinRestartBackoff } r.log.Warn("ingressproxy_opentunnel_exited", "err", fmt.Sprint(err), "output", stderr) if !clock.Sleep(ctx, backoff) { return } backoff = childproc.NextBackoff(backoff, MinRestartBackoff, MaxRestartBackoff) } } func pump(reader io.Reader, tail *childproc.Tail, found chan<- string) { scanner := bufio.NewScanner(reader) scanner.Buffer(make([]byte, 4096), 64*1024) for scanner.Scan() { line := scanner.Text() tail.Write([]byte(line + "\n")) if raw, ok := urlIn(line); ok { select { case found <- raw: default: } } } } func urlIn(line string) (string, bool) { for _, match := range urlPattern.FindAllString(line, -1) { candidate := strings.TrimRight(match, ".,;:)]}") parsed, err := url.Parse(candidate) if err != nil || parsed.Host == "" { continue } host := strings.ToLower(parsed.Hostname()) if host != "opentunnel.xyz" && strings.HasSuffix(host, ".opentunnel.xyz") && strings.Count(host, ".") >= 2 { return candidate, true } } return "", false }