fix(prose,pipe): serve a post named raw and bound pipe-web connection setup

Commit `def8a10`

Antonio Mika · 2026-10-10 23:43:30 -0400 EDT

parent 08f4e64

fix(prose,pipe): serve a post named raw and bound pipe-web connection setup

5 files changed, +421, -27

M pkg/apps/pipe/api.go

M pkg/apps/pipe/conn.go

A pkg/apps/pipe/conn_test.go

M pkg/apps/prose/api.go

M pkg/apps/prose/api_test.go

M pkg/apps/pipe/api.go

+3, -3

......@@ -76,7 +76,7 @@ func (web *webRouter) handleSub(pubsub bool) http.HandlerFunc {
			params += fmt.Sprintf(" -a=%s", cleanList)
		}

		p, err := web.conn.run(fmt.Sprintf("sub %s %s", params, topic))
		p, err := web.conn.run(r.Context(), fmt.Sprintf("sub %s %s", params, topic))
		if err != nil {
			logger.Error("sub error", "topic", topic, "info", clientInfo, "err", err.Error())
			http.Error(w, "server error", http.StatusInternalServerError)

......@@ -141,7 +141,7 @@ func (web *webRouter) handlePub(pubsub bool) http.HandlerFunc {
			params += " -e"
		}

		p, err := web.conn.run(fmt.Sprintf("pub %s %s", params, topic))
		p, err := web.conn.run(r.Context(), fmt.Sprintf("pub %s %s", params, topic))
		if err != nil {
			logger.Error("pub error", "topic", topic, "info", clientInfo, "err", err.Error())
			http.Error(w, "server error", http.StatusInternalServerError)

......@@ -273,7 +273,7 @@ func (web *webRouter) handlePipe() http.HandlerFunc {

		prefix := r.URL.Query().Get("prefix")

		p, err := web.conn.run(fmt.Sprintf("pipe %s %s", params, topic))
		p, err := web.conn.run(r.Context(), fmt.Sprintf("pipe %s %s", params, topic))
		if err != nil {
			logger.Error("pipe error", "topic", topic, "info", clientInfo, "err", err.Error())
			_ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "server error"), time.Now().Add(time.Second))

M pkg/apps/pipe/conn.go

+176, -19

......@@ -1,63 +1,220 @@
package pipe

import (
	"context"
	"fmt"
	"io"
	"net"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/picosh/utils/pipe"
	"golang.org/x/crypto/ssh"
)

// dialTimeout bounds connecting to pipe, handshake included, and starting a
// command on it, so a server that stops answering can't hold up requests.
const dialTimeout = 10 * time.Second

// pipeConn is pipe-web's ssh connection to pipe. Each request runs its
// command in a session of its own, which is never reopened, so the command
// ends when the request does.
type pipeConn struct {
	info *pipe.SSHClientInfo
	info    *pipe.SSHClientInfo
	timeout time.Duration

	mu      sync.Mutex
	client  *ssh.Client
	dialing *dial
}

	mu     sync.Mutex
// dial is a connection being opened, shared by the requests waiting on it.
type dial struct {
	done   chan struct{}
	client *ssh.Client
	err    error
}

func newPipeConn(info *pipe.SSHClientInfo) *pipeConn {
	return &pipeConn{info: info}
	return &pipeConn{info: info, timeout: dialTimeout}
}

// connect returns the open connection, dialing a new one when there is none
// or when the open one is dead.
func (c *pipeConn) connect(dead *ssh.Client) (*ssh.Client, error) {
// or when the open one is dead. A request that ends stops waiting, and the
// dial carries on for the requests after it.
func (c *pipeConn) connect(ctx context.Context, dead *ssh.Client) (*ssh.Client, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.client != nil && c.client != dead {
		return c.client, nil
		client := c.client
		c.mu.Unlock()
		return client, nil
	}
	if c.client != nil {
		_ = c.client.Close()
		c.client = nil
	}
	client, err := pipe.NewSSHClient(c.info)
	if err != nil {
		return nil, err
	d := c.dialing
	if d == nil {
		d = &dial{done: make(chan struct{})}
		c.dialing = d
		go c.dial(d)
	}
	c.mu.Unlock()

	select {
	case <-d.done:
		return d.client, d.err
	case <-ctx.Done():
		return nil, ctx.Err()
	}
	c.client = client
	return client, nil
}

func (c *pipeConn) dial(d *dial) {
	d.client, d.err = dialSSH(c.info, c.timeout)
	c.mu.Lock()
	c.dialing = nil
	if d.err == nil {
		c.client = d.client
	}
	c.mu.Unlock()
	close(d.done)
}

// run starts cmd on pipe. A connection that can't open a session is dialed
// again once.
func (c *pipeConn) run(cmd string) (*cmdSession, error) {
	client, err := c.connect(nil)
func (c *pipeConn) run(ctx context.Context, cmd string) (*cmdSession, error) {
	client, err := c.connect(ctx, nil)
	if err != nil {
		return nil, err
	}
	s, err := startCmd(client, cmd)
	if err == nil {
		return s, nil
	s, err := c.start(ctx, client, cmd)
	if err == nil || ctx.Err() != nil {
		return s, err
	}
	client, err = c.connect(client)
	client, err = c.connect(ctx, client)
	if err != nil {
		return nil, err
	}
	return startCmd(client, cmd)
	return c.start(ctx, client, cmd)
}

type started struct {
	s   *cmdSession
	err error
}

// start runs startCmd until the request ends. A connection that can't
// start a command in time is closed, which also ends the startCmd waiting
// on it, and that deadline holds after the request has gone.
func (c *pipeConn) start(ctx context.Context, client *ssh.Client, cmd string) (*cmdSession, error) {
	done := make(chan started, 1)
	go func() {
		s, err := startCmd(client, cmd)
		done <- started{s, err}
	}()
	timer := time.NewTimer(c.timeout)
	select {
	case res := <-done:
		timer.Stop()
		return res.s, res.err
	case <-ctx.Done():
		go abandon(done, timer, client)
		return nil, ctx.Err()
	case <-timer.C:
		_ = client.Close()
		go closeStarted(done)
		return nil, fmt.Errorf("starting %q on pipe timed out", cmd)
	}
}

// abandon closes a session no request is waiting for, or the connection if
// the session doesn't start in time.
func abandon(done <-chan started, timer *time.Timer, client *ssh.Client) {
	select {
	case res := <-done:
		timer.Stop()
		if res.s != nil {
			_ = res.s.Close()
		}
	case <-timer.C:
		_ = client.Close()
		closeStarted(done)
	}
}

func closeStarted(done <-chan started) {
	if res := <-done; res.s != nil {
		_ = res.s.Close()
	}
}

func dialSSH(info *pipe.SSHClientInfo, timeout time.Duration) (*ssh.Client, error) {
	signer, err := clientSigner(info)
	if err != nil {
		return nil, err
	}
	var auth []ssh.AuthMethod
	if signer != nil {
		auth = append(auth, ssh.PublicKeys(signer))
	}
	addr := info.RemoteHost
	if !strings.Contains(addr, ":") {
		addr += ":22"
	}

	deadline := time.Now().Add(timeout)
	conn, err := (&net.Dialer{Deadline: deadline}).Dial("tcp", addr)
	if err != nil {
		return nil, err
	}
	_ = conn.SetDeadline(deadline)
	sshConn, chans, reqs, err := ssh.NewClientConn(conn, info.RemoteHostname, &ssh.ClientConfig{
		Auth:            auth,
		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
		User:            info.RemoteUser,
	})
	if err != nil {
		_ = conn.Close()
		return nil, err
	}
	_ = conn.SetDeadline(time.Time{})
	return ssh.NewClient(sshConn, chans, reqs), nil
}

// clientSigner loads the key, and the certificate for it if there is one.
// It returns nil when there is no key.
func clientSigner(info *pipe.SSHClientInfo) (ssh.Signer, error) {
	if info.KeyLocation == "" {
		return nil, nil
	}
	data, err := os.ReadFile(info.KeyLocation)
	if err != nil {
		return nil, err
	}
	var signer ssh.Signer
	if info.KeyPassphrase != "" {
		signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(info.KeyPassphrase))
	} else {
		signer, err = ssh.ParsePrivateKey(data)
	}
	if err != nil || info.CertificateLocation == "" {
		return signer, err
	}
	certData, err := os.ReadFile(info.CertificateLocation)
	if err != nil {
		return nil, err
	}
	pub, _, _, _, err := ssh.ParseAuthorizedKey(certData)
	if err != nil {
		return nil, fmt.Errorf("failed to parse certificate: %w", err)
	}
	cert, ok := pub.(*ssh.Certificate)
	if !ok {
		return nil, fmt.Errorf("%s is not an ssh certificate", info.CertificateLocation)
	}
	return ssh.NewCertSigner(cert, signer)
}

// cmdSession is a command running on pipe, read from its stdout and written

A pkg/apps/pipe/conn_test.go

+227, -0

......@@ -0,0 +1,227 @@
package pipe

import (
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"errors"
	"io"
	"net"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/picosh/utils/pipe"
	"golang.org/x/crypto/ssh"
)

// closers closes what a test server opened once the test ends.
type closers struct {
	mu sync.Mutex
	cs []io.Closer
}

func newClosers(t *testing.T) *closers {
	c := &closers{}
	t.Cleanup(func() {
		c.mu.Lock()
		defer c.mu.Unlock()
		for _, cl := range c.cs {
			_ = cl.Close()
		}
	})
	return c
}

func (c *closers) add(cl io.Closer) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.cs = append(c.cs, cl)
}

// stalledConn is a pipe connection to a server that accepts connections and
// never answers.
func stalledConn(t *testing.T, timeout time.Duration) *pipeConn {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	open := newClosers(t)
	open.add(ln)
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			open.add(conn)
		}
	}()
	conn := newPipeConn(&pipe.SSHClientInfo{RemoteHost: ln.Addr().String()})
	conn.timeout = timeout
	return conn
}

func TestStalledHandshakeEndsWithRequest(t *testing.T) {
	conn := stalledConn(t, time.Minute)

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	start := time.Now()
	if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("run = %v, want the request's deadline", err)
	}
	if took := time.Since(start); took > time.Second {
		t.Errorf("canceled request waited %v for the handshake", took)
	}

	// The next request is not held up behind the canceled one.
	ctx, cancel = context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	start = time.Now()
	if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("run = %v, want the request's deadline", err)
	}
	if took := time.Since(start); took > time.Second {
		t.Errorf("second request waited %v", took)
	}
}

func TestStalledHandshakeTimesOut(t *testing.T) {
	conn := stalledConn(t, 200*time.Millisecond)

	start := time.Now()
	_, err := conn.run(context.Background(), "sub topic")
	var netErr net.Error
	if !errors.As(err, &netErr) || !netErr.Timeout() {
		t.Fatalf("run = %v, want a timeout", err)
	}
	if took := time.Since(start); took > 2*time.Second {
		t.Errorf("handshake took %v, want about 200ms", took)
	}
}

// stalledSSH is a pipe connection to an ssh server that completes the
// handshake and then stops answering: at opening a session when openStalls
// is set, or else at starting the command. It counts the connections made
// and the ones closed.
func stalledSSH(t *testing.T, timeout time.Duration, openStalls bool) (conn *pipeConn, conns, closed *atomic.Int32) {
	t.Helper()
	_, key, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	hostKey, err := ssh.NewSignerFromKey(key)
	if err != nil {
		t.Fatal(err)
	}
	cfg := &ssh.ServerConfig{NoClientAuth: true}
	cfg.AddHostKey(hostKey)

	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	open := newClosers(t)
	open.add(ln)
	conns, closed = &atomic.Int32{}, &atomic.Int32{}
	go func() {
		for {
			nc, err := ln.Accept()
			if err != nil {
				return
			}
			conns.Add(1)
			open.add(nc)
			go func() {
				sc, chans, reqs, err := ssh.NewServerConn(nc, cfg)
				if err != nil {
					return
				}
				go func() {
					_ = sc.Wait()
					closed.Add(1)
				}()
				go ssh.DiscardRequests(reqs)
				for nch := range chans {
					if openStalls {
						continue
					}
					ch, chReqs, err := nch.Accept()
					if err != nil {
						return
					}
					open.add(ch)
					// Never reply to the exec request.
					go func() {
						for range chReqs {
						}
					}()
				}
			}()
		}
	}()
	conn = newPipeConn(&pipe.SSHClientInfo{RemoteHost: ln.Addr().String()})
	conn.timeout = timeout
	return conn, conns, closed
}

func TestStalledSessionEndsWithRequest(t *testing.T) {
	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
		t.Run(name, func(t *testing.T) {
			conn, _, _ := stalledSSH(t, time.Minute, openStalls)
			ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
			defer cancel()
			start := time.Now()
			if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
				t.Fatalf("run = %v, want the request's deadline", err)
			}
			if took := time.Since(start); took > time.Second {
				t.Errorf("canceled request waited %v for the session", took)
			}
		})
	}
}

func TestStalledSessionRedials(t *testing.T) {
	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
		t.Run(name, func(t *testing.T) {
			conn, conns, _ := stalledSSH(t, 200*time.Millisecond, openStalls)
			_, err := conn.run(context.Background(), "sub topic")
			if err == nil || !strings.Contains(err.Error(), "timed out") {
				t.Fatalf("run = %v, want a timeout", err)
			}
			if n := conns.Load(); n != 2 {
				t.Errorf("made %d connections, want the stalled one and a redial", n)
			}
		})
	}
}

func TestCanceledStartStillTimesOut(t *testing.T) {
	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
		t.Run(name, func(t *testing.T) {
			conn, _, closed := stalledSSH(t, 300*time.Millisecond, openStalls)
			before := runtime.NumGoroutine()
			for range 5 {
				ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
				_, err := conn.run(ctx, "sub topic")
				cancel()
				if !errors.Is(err, context.DeadlineExceeded) {
					t.Fatalf("run = %v, want the request's deadline", err)
				}
			}
			for range 100 {
				if closed.Load() == 1 && runtime.NumGoroutine() <= before {
					return
				}
				time.Sleep(20 * time.Millisecond)
			}
			t.Errorf("closed %d connections and have %d goroutines, want 1 and at most %d", closed.Load(), runtime.NumGoroutine(), before)
		})
	}
}

M pkg/apps/prose/api.go

+6, -1

......@@ -922,7 +922,6 @@ func (web *webRouter) sitesMux() *http.ServeMux {
		mux.HandleFunc("GET "+feed, web.rssBlogHandler)
	}
	web.static(mux)
	mux.HandleFunc("GET /raw/{slug...}", web.postRawHandler)
	mux.HandleFunc("GET /{path...}", web.pathRoute)
	return mux
}

......@@ -934,8 +933,14 @@ var (

// pathRoute serves the site paths ServeMux patterns can't tell apart: a
// post's markdown, an image with or without processing options, and a post.
// A "/raw/{slug...}" pattern would redirect "/raw" away from a post named raw.
func (web *webRouter) pathRoute(w http.ResponseWriter, r *http.Request) {
	path := r.PathValue("path")
	if slug, ok := strings.CutPrefix(path, "raw/"); ok && slug != "" {
		r.SetPathValue("slug", slug)
		web.postRawHandler(w, r)
		return
	}
	for _, ext := range []string{".md", ".lxt"} {
		if slug, ok := strings.CutSuffix(path, ext); ok && slug != "" {
			r.SetPathValue("slug", slug)

M pkg/apps/prose/api_test.go

+9, -4

......@@ -22,10 +22,12 @@ func TestRoutes(t *testing.T) {
	posts := stub.NewPostsDB(logger, user)
	cfg := NewConfig("prose-test")
	cfg.Logger = logger
	post := filehandlers.NewPost("hello.md", []byte("# hello"))
	post.UserID, post.Space, post.Title = user.ID, cfg.Space, "Hello"
	if _, err := posts.InsertPost(post); err != nil {
		t.Fatal(err)
	for _, name := range []string{"hello", "raw"} {
		post := filehandlers.NewPost(name+".md", []byte("# "+name))
		post.UserID, post.Space, post.Title = user.ID, cfg.Space, name
		if _, err := posts.InsertPost(post); err != nil {
			t.Fatal(err)
		}
	}
	st, err := storage.NewStorageMemory(map[string]map[string]string{})
	if err != nil {

......@@ -48,6 +50,9 @@ func TestRoutes(t *testing.T) {
		{url: site + "/hello.md", status: http.StatusOK, body: "# hello"},
		{url: site + "/hello.lxt", status: http.StatusOK, body: "# hello"},
		{url: site + "/raw/hello", status: http.StatusOK, body: "# hello"},
		{url: site + "/raw", status: http.StatusOK, body: "<html"},
		{url: site + "/raw.md", status: http.StatusOK, body: "# raw"},
		{url: site + "/raw/raw", status: http.StatusOK, body: "# raw"},
		{url: site + "/missing", status: http.StatusNotFound},
		// Images are served from the user's bucket, which they don't have.
		{url: site + "/cat.png", status: http.StatusUnprocessableEntity, body: "bucket does not exist"},

...	...		@@ -76,7 +76,7 @@ func (web *webRouter) handleSub(pubsub bool) http.HandlerFunc {
76	76		params += fmt.Sprintf(" -a=%s", cleanList)
77	77		}
78	78
79		-	p, err := web.conn.run(fmt.Sprintf("sub %s %s", params, topic))
	79	+	p, err := web.conn.run(r.Context(), fmt.Sprintf("sub %s %s", params, topic))
80	80		if err != nil {
81	81		logger.Error("sub error", "topic", topic, "info", clientInfo, "err", err.Error())
82	82		http.Error(w, "server error", http.StatusInternalServerError)

...	...		@@ -141,7 +141,7 @@ func (web *webRouter) handlePub(pubsub bool) http.HandlerFunc {
141	141		params += " -e"
142	142		}
143	143
144		-	p, err := web.conn.run(fmt.Sprintf("pub %s %s", params, topic))
	144	+	p, err := web.conn.run(r.Context(), fmt.Sprintf("pub %s %s", params, topic))
145	145		if err != nil {
146	146		logger.Error("pub error", "topic", topic, "info", clientInfo, "err", err.Error())
147	147		http.Error(w, "server error", http.StatusInternalServerError)

...	...		@@ -273,7 +273,7 @@ func (web *webRouter) handlePipe() http.HandlerFunc {
273	273
274	274		prefix := r.URL.Query().Get("prefix")
275	275
276		-	p, err := web.conn.run(fmt.Sprintf("pipe %s %s", params, topic))
	276	+	p, err := web.conn.run(r.Context(), fmt.Sprintf("pipe %s %s", params, topic))
277	277		if err != nil {
278	278		logger.Error("pipe error", "topic", topic, "info", clientInfo, "err", err.Error())
279	279		_ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "server error"), time.Now().Add(time.Second))

...	...		@@ -922,7 +922,6 @@ func (web webRouter) sitesMux() http.ServeMux {
922	922		mux.HandleFunc("GET "+feed, web.rssBlogHandler)
923	923		}
924	924		web.static(mux)
925		-	mux.HandleFunc("GET /raw/{slug...}", web.postRawHandler)
926	925		mux.HandleFunc("GET /{path...}", web.pathRoute)
927	926		return mux
928	927		}

...	...		@@ -934,8 +933,14 @@ var (
934	933
935	934		// pathRoute serves the site paths ServeMux patterns can't tell apart: a
936	935		// post's markdown, an image with or without processing options, and a post.
	936	+	// A "/raw/{slug...}" pattern would redirect "/raw" away from a post named raw.
937	937		func (web webRouter) pathRoute(w http.ResponseWriter, r http.Request) {
938	938		path := r.PathValue("path")
	939	+	if slug, ok := strings.CutPrefix(path, "raw/"); ok && slug != "" {
	940	+	r.SetPathValue("slug", slug)
	941	+	web.postRawHandler(w, r)
	942	+	return
	943	+	}
939	944		for _, ext := range []string{".md", ".lxt"} {
940	945		if slug, ok := strings.CutSuffix(path, ext); ok && slug != "" {
941	946		r.SetPathValue("slug", slug)

...	...		@@ -1,63 +1,220 @@
1	1		package pipe
2	2
3	3		import (
	4	+	"context"
	5	+	"fmt"
4	6		"io"
	7	+	"net"
	8	+	"os"
	9	+	"strings"
5	10		"sync"
	11	+	"time"
6	12
7	13		"github.com/picosh/utils/pipe"
8	14		"golang.org/x/crypto/ssh"
9	15		)
10	16
	17	+	// dialTimeout bounds connecting to pipe, handshake included, and starting a
	18	+	// command on it, so a server that stops answering can't hold up requests.
	19	+	const dialTimeout = 10 * time.Second
	20	+
11	21		// pipeConn is pipe-web's ssh connection to pipe. Each request runs its
12	22		// command in a session of its own, which is never reopened, so the command
13	23		// ends when the request does.
14	24		type pipeConn struct {
15		-	info *pipe.SSHClientInfo
	25	+	info *pipe.SSHClientInfo
	26	+	timeout time.Duration
	27	+
	28	+	mu sync.Mutex
	29	+	client *ssh.Client
	30	+	dialing *dial
	31	+	}
16	32
17		-	mu sync.Mutex
	33	+	// dial is a connection being opened, shared by the requests waiting on it.
	34	+	type dial struct {
	35	+	done chan struct{}
18	36		client *ssh.Client
	37	+	err error
19	38		}
20	39
21	40		func newPipeConn(info pipe.SSHClientInfo) pipeConn {
22		-	return &pipeConn{info: info}
	41	+	return &pipeConn{info: info, timeout: dialTimeout}
23	42		}
24	43
25	44		// connect returns the open connection, dialing a new one when there is none
26		-	// or when the open one is dead.
27		-	func (c pipeConn) connect(dead ssh.Client) (*ssh.Client, error) {
	45	+	// or when the open one is dead. A request that ends stops waiting, and the
	46	+	// dial carries on for the requests after it.
	47	+	func (c pipeConn) connect(ctx context.Context, dead ssh.Client) (*ssh.Client, error) {
28	48		c.mu.Lock()
29		-	defer c.mu.Unlock()
30	49		if c.client != nil && c.client != dead {
31		-	return c.client, nil
	50	+	client := c.client
	51	+	c.mu.Unlock()
	52	+	return client, nil
32	53		}
33	54		if c.client != nil {
34	55		_ = c.client.Close()
35	56		c.client = nil
36	57		}
37		-	client, err := pipe.NewSSHClient(c.info)
38		-	if err != nil {
39		-	return nil, err
	58	+	d := c.dialing
	59	+	if d == nil {
	60	+	d = &dial{done: make(chan struct{})}
	61	+	c.dialing = d
	62	+	go c.dial(d)
	63	+	}
	64	+	c.mu.Unlock()
	65	+
	66	+	select {
	67	+	case <-d.done:
	68	+	return d.client, d.err
	69	+	case <-ctx.Done():
	70	+	return nil, ctx.Err()
40	71		}
41		-	c.client = client
42		-	return client, nil
	72	+	}
	73	+
	74	+	func (c pipeConn) dial(d dial) {
	75	+	d.client, d.err = dialSSH(c.info, c.timeout)
	76	+	c.mu.Lock()
	77	+	c.dialing = nil
	78	+	if d.err == nil {
	79	+	c.client = d.client
	80	+	}
	81	+	c.mu.Unlock()
	82	+	close(d.done)
43	83		}
44	84
45	85		// run starts cmd on pipe. A connection that can't open a session is dialed
46	86		// again once.
47		-	func (c pipeConn) run(cmd string) (cmdSession, error) {
48		-	client, err := c.connect(nil)
	87	+	func (c pipeConn) run(ctx context.Context, cmd string) (cmdSession, error) {
	88	+	client, err := c.connect(ctx, nil)
49	89		if err != nil {
50	90		return nil, err
51	91		}
52		-	s, err := startCmd(client, cmd)
53		-	if err == nil {
54		-	return s, nil
	92	+	s, err := c.start(ctx, client, cmd)
	93	+	if err == nil \|\| ctx.Err() != nil {
	94	+	return s, err
55	95		}
56		-	client, err = c.connect(client)
	96	+	client, err = c.connect(ctx, client)
57	97		if err != nil {
58	98		return nil, err
59	99		}
60		-	return startCmd(client, cmd)
	100	+	return c.start(ctx, client, cmd)
	101	+	}
	102	+
	103	+	type started struct {
	104	+	s *cmdSession
	105	+	err error
	106	+	}
	107	+
	108	+	// start runs startCmd until the request ends. A connection that can't
	109	+	// start a command in time is closed, which also ends the startCmd waiting
	110	+	// on it, and that deadline holds after the request has gone.
	111	+	func (c pipeConn) start(ctx context.Context, client ssh.Client, cmd string) (*cmdSession, error) {
	112	+	done := make(chan started, 1)
	113	+	go func() {
	114	+	s, err := startCmd(client, cmd)
	115	+	done <- started{s, err}
	116	+	}()
	117	+	timer := time.NewTimer(c.timeout)
	118	+	select {
	119	+	case res := <-done:
	120	+	timer.Stop()
	121	+	return res.s, res.err
	122	+	case <-ctx.Done():
	123	+	go abandon(done, timer, client)
	124	+	return nil, ctx.Err()
	125	+	case <-timer.C:
	126	+	_ = client.Close()
	127	+	go closeStarted(done)
	128	+	return nil, fmt.Errorf("starting %q on pipe timed out", cmd)
	129	+	}
	130	+	}
	131	+
	132	+	// abandon closes a session no request is waiting for, or the connection if
	133	+	// the session doesn't start in time.
	134	+	func abandon(done <-chan started, timer time.Timer, client ssh.Client) {
	135	+	select {
	136	+	case res := <-done:
	137	+	timer.Stop()
	138	+	if res.s != nil {
	139	+	_ = res.s.Close()
	140	+	}
	141	+	case <-timer.C:
	142	+	_ = client.Close()
	143	+	closeStarted(done)
	144	+	}
	145	+	}
	146	+
	147	+	func closeStarted(done <-chan started) {
	148	+	if res := <-done; res.s != nil {
	149	+	_ = res.s.Close()
	150	+	}
	151	+	}
	152	+
	153	+	func dialSSH(info pipe.SSHClientInfo, timeout time.Duration) (ssh.Client, error) {
	154	+	signer, err := clientSigner(info)
	155	+	if err != nil {
	156	+	return nil, err
	157	+	}
	158	+	var auth []ssh.AuthMethod
	159	+	if signer != nil {
	160	+	auth = append(auth, ssh.PublicKeys(signer))
	161	+	}
	162	+	addr := info.RemoteHost
	163	+	if !strings.Contains(addr, ":") {
	164	+	addr += ":22"
	165	+	}
	166	+
	167	+	deadline := time.Now().Add(timeout)
	168	+	conn, err := (&net.Dialer{Deadline: deadline}).Dial("tcp", addr)
	169	+	if err != nil {
	170	+	return nil, err
	171	+	}
	172	+	_ = conn.SetDeadline(deadline)
	173	+	sshConn, chans, reqs, err := ssh.NewClientConn(conn, info.RemoteHostname, &ssh.ClientConfig{
	174	+	Auth: auth,
	175	+	HostKeyCallback: ssh.InsecureIgnoreHostKey(),
	176	+	User: info.RemoteUser,
	177	+	})
	178	+	if err != nil {
	179	+	_ = conn.Close()
	180	+	return nil, err
	181	+	}
	182	+	_ = conn.SetDeadline(time.Time{})
	183	+	return ssh.NewClient(sshConn, chans, reqs), nil
	184	+	}
	185	+
	186	+	// clientSigner loads the key, and the certificate for it if there is one.
	187	+	// It returns nil when there is no key.
	188	+	func clientSigner(info *pipe.SSHClientInfo) (ssh.Signer, error) {
	189	+	if info.KeyLocation == "" {
	190	+	return nil, nil
	191	+	}
	192	+	data, err := os.ReadFile(info.KeyLocation)
	193	+	if err != nil {
	194	+	return nil, err
	195	+	}
	196	+	var signer ssh.Signer
	197	+	if info.KeyPassphrase != "" {
	198	+	signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(info.KeyPassphrase))
	199	+	} else {
	200	+	signer, err = ssh.ParsePrivateKey(data)
	201	+	}
	202	+	if err != nil \|\| info.CertificateLocation == "" {
	203	+	return signer, err
	204	+	}
	205	+	certData, err := os.ReadFile(info.CertificateLocation)
	206	+	if err != nil {
	207	+	return nil, err
	208	+	}
	209	+	pub, _, _, _, err := ssh.ParseAuthorizedKey(certData)
	210	+	if err != nil {
	211	+	return nil, fmt.Errorf("failed to parse certificate: %w", err)
	212	+	}
	213	+	cert, ok := pub.(*ssh.Certificate)
	214	+	if !ok {
	215	+	return nil, fmt.Errorf("%s is not an ssh certificate", info.CertificateLocation)
	216	+	}
	217	+	return ssh.NewCertSigner(cert, signer)
61	218		}
62	219
63	220		// cmdSession is a command running on pipe, read from its stdout and written

...	...		@@ -0,0 +1,227 @@
	1	+	package pipe
	2	+
	3	+	import (
	4	+	"context"
	5	+	"crypto/ed25519"
	6	+	"crypto/rand"
	7	+	"errors"
	8	+	"io"
	9	+	"net"
	10	+	"runtime"
	11	+	"strings"
	12	+	"sync"
	13	+	"sync/atomic"
	14	+	"testing"
	15	+	"time"
	16	+
	17	+	"github.com/picosh/utils/pipe"
	18	+	"golang.org/x/crypto/ssh"
	19	+	)
	20	+
	21	+	// closers closes what a test server opened once the test ends.
	22	+	type closers struct {
	23	+	mu sync.Mutex
	24	+	cs []io.Closer
	25	+	}
	26	+
	27	+	func newClosers(t testing.T) closers {
	28	+	c := &closers{}
	29	+	t.Cleanup(func() {
	30	+	c.mu.Lock()
	31	+	defer c.mu.Unlock()
	32	+	for _, cl := range c.cs {
	33	+	_ = cl.Close()
	34	+	}
	35	+	})
	36	+	return c
	37	+	}
	38	+
	39	+	func (c *closers) add(cl io.Closer) {
	40	+	c.mu.Lock()
	41	+	defer c.mu.Unlock()
	42	+	c.cs = append(c.cs, cl)
	43	+	}
	44	+
	45	+	// stalledConn is a pipe connection to a server that accepts connections and
	46	+	// never answers.
	47	+	func stalledConn(t testing.T, timeout time.Duration) pipeConn {
	48	+	t.Helper()
	49	+	ln, err := net.Listen("tcp", "127.0.0.1:0")
	50	+	if err != nil {
	51	+	t.Fatal(err)
	52	+	}
	53	+	open := newClosers(t)
	54	+	open.add(ln)
	55	+	go func() {
	56	+	for {
	57	+	conn, err := ln.Accept()
	58	+	if err != nil {
	59	+	return
	60	+	}
	61	+	open.add(conn)
	62	+	}
	63	+	}()
	64	+	conn := newPipeConn(&pipe.SSHClientInfo{RemoteHost: ln.Addr().String()})
	65	+	conn.timeout = timeout
	66	+	return conn
	67	+	}
	68	+
	69	+	func TestStalledHandshakeEndsWithRequest(t *testing.T) {
	70	+	conn := stalledConn(t, time.Minute)
	71	+
	72	+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	73	+	defer cancel()
	74	+	start := time.Now()
	75	+	if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
	76	+	t.Fatalf("run = %v, want the request's deadline", err)
	77	+	}
	78	+	if took := time.Since(start); took > time.Second {
	79	+	t.Errorf("canceled request waited %v for the handshake", took)
	80	+	}
	81	+
	82	+	// The next request is not held up behind the canceled one.
	83	+	ctx, cancel = context.WithTimeout(context.Background(), 100*time.Millisecond)
	84	+	defer cancel()
	85	+	start = time.Now()
	86	+	if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
	87	+	t.Fatalf("run = %v, want the request's deadline", err)
	88	+	}
	89	+	if took := time.Since(start); took > time.Second {
	90	+	t.Errorf("second request waited %v", took)
	91	+	}
	92	+	}
	93	+
	94	+	func TestStalledHandshakeTimesOut(t *testing.T) {
	95	+	conn := stalledConn(t, 200*time.Millisecond)
	96	+
	97	+	start := time.Now()
	98	+	_, err := conn.run(context.Background(), "sub topic")
	99	+	var netErr net.Error
	100	+	if !errors.As(err, &netErr) \|\| !netErr.Timeout() {
	101	+	t.Fatalf("run = %v, want a timeout", err)
	102	+	}
	103	+	if took := time.Since(start); took > 2*time.Second {
	104	+	t.Errorf("handshake took %v, want about 200ms", took)
	105	+	}
	106	+	}
	107	+
	108	+	// stalledSSH is a pipe connection to an ssh server that completes the
	109	+	// handshake and then stops answering: at opening a session when openStalls
	110	+	// is set, or else at starting the command. It counts the connections made
	111	+	// and the ones closed.
	112	+	func stalledSSH(t testing.T, timeout time.Duration, openStalls bool) (conn pipeConn, conns, closed *atomic.Int32) {
	113	+	t.Helper()
	114	+	_, key, err := ed25519.GenerateKey(rand.Reader)
	115	+	if err != nil {
	116	+	t.Fatal(err)
	117	+	}
	118	+	hostKey, err := ssh.NewSignerFromKey(key)
	119	+	if err != nil {
	120	+	t.Fatal(err)
	121	+	}
	122	+	cfg := &ssh.ServerConfig{NoClientAuth: true}
	123	+	cfg.AddHostKey(hostKey)
	124	+
	125	+	ln, err := net.Listen("tcp", "127.0.0.1:0")
	126	+	if err != nil {
	127	+	t.Fatal(err)
	128	+	}
	129	+	open := newClosers(t)
	130	+	open.add(ln)
	131	+	conns, closed = &atomic.Int32{}, &atomic.Int32{}
	132	+	go func() {
	133	+	for {
	134	+	nc, err := ln.Accept()
	135	+	if err != nil {
	136	+	return
	137	+	}
	138	+	conns.Add(1)
	139	+	open.add(nc)
	140	+	go func() {
	141	+	sc, chans, reqs, err := ssh.NewServerConn(nc, cfg)
	142	+	if err != nil {
	143	+	return
	144	+	}
	145	+	go func() {
	146	+	_ = sc.Wait()
	147	+	closed.Add(1)
	148	+	}()
	149	+	go ssh.DiscardRequests(reqs)
	150	+	for nch := range chans {
	151	+	if openStalls {
	152	+	continue
	153	+	}
	154	+	ch, chReqs, err := nch.Accept()
	155	+	if err != nil {
	156	+	return
	157	+	}
	158	+	open.add(ch)
	159	+	// Never reply to the exec request.
	160	+	go func() {
	161	+	for range chReqs {
	162	+	}
	163	+	}()
	164	+	}
	165	+	}()
	166	+	}
	167	+	}()
	168	+	conn = newPipeConn(&pipe.SSHClientInfo{RemoteHost: ln.Addr().String()})
	169	+	conn.timeout = timeout
	170	+	return conn, conns, closed
	171	+	}
	172	+
	173	+	func TestStalledSessionEndsWithRequest(t *testing.T) {
	174	+	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
	175	+	t.Run(name, func(t *testing.T) {
	176	+	conn, _, _ := stalledSSH(t, time.Minute, openStalls)
	177	+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
	178	+	defer cancel()
	179	+	start := time.Now()
	180	+	if _, err := conn.run(ctx, "sub topic"); !errors.Is(err, context.DeadlineExceeded) {
	181	+	t.Fatalf("run = %v, want the request's deadline", err)
	182	+	}
	183	+	if took := time.Since(start); took > time.Second {
	184	+	t.Errorf("canceled request waited %v for the session", took)
	185	+	}
	186	+	})
	187	+	}
	188	+	}
	189	+
	190	+	func TestStalledSessionRedials(t *testing.T) {
	191	+	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
	192	+	t.Run(name, func(t *testing.T) {
	193	+	conn, conns, _ := stalledSSH(t, 200*time.Millisecond, openStalls)
	194	+	_, err := conn.run(context.Background(), "sub topic")
	195	+	if err == nil \|\| !strings.Contains(err.Error(), "timed out") {
	196	+	t.Fatalf("run = %v, want a timeout", err)
	197	+	}
	198	+	if n := conns.Load(); n != 2 {
	199	+	t.Errorf("made %d connections, want the stalled one and a redial", n)
	200	+	}
	201	+	})
	202	+	}
	203	+	}
	204	+
	205	+	func TestCanceledStartStillTimesOut(t *testing.T) {
	206	+	for name, openStalls := range map[string]bool{"open": true, "exec": false} {
	207	+	t.Run(name, func(t *testing.T) {
	208	+	conn, _, closed := stalledSSH(t, 300*time.Millisecond, openStalls)
	209	+	before := runtime.NumGoroutine()
	210	+	for range 5 {
	211	+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	212	+	_, err := conn.run(ctx, "sub topic")
	213	+	cancel()
	214	+	if !errors.Is(err, context.DeadlineExceeded) {
	215	+	t.Fatalf("run = %v, want the request's deadline", err)
	216	+	}
	217	+	}
	218	+	for range 100 {
	219	+	if closed.Load() == 1 && runtime.NumGoroutine() <= before {
	220	+	return
	221	+	}
	222	+	time.Sleep(20 * time.Millisecond)
	223	+	}
	224	+	t.Errorf("closed %d connections and have %d goroutines, want 1 and at most %d", closed.Load(), runtime.NumGoroutine(), before)
	225	+	})
	226	+	}
	227	+	}

...	...		@@ -22,10 +22,12 @@ func TestRoutes(t *testing.T) {
22	22		posts := stub.NewPostsDB(logger, user)
23	23		cfg := NewConfig("prose-test")
24	24		cfg.Logger = logger
25		-	post := filehandlers.NewPost("hello.md", []byte("# hello"))
26		-	post.UserID, post.Space, post.Title = user.ID, cfg.Space, "Hello"
27		-	if _, err := posts.InsertPost(post); err != nil {
28		-	t.Fatal(err)
	25	+	for _, name := range []string{"hello", "raw"} {
	26	+	post := filehandlers.NewPost(name+".md", []byte("# "+name))
	27	+	post.UserID, post.Space, post.Title = user.ID, cfg.Space, name
	28	+	if _, err := posts.InsertPost(post); err != nil {
	29	+	t.Fatal(err)
	30	+	}
29	31		}
30	32		st, err := storage.NewStorageMemory(map[string]map[string]string{})
31	33		if err != nil {

...	...		@@ -48,6 +50,9 @@ func TestRoutes(t *testing.T) {
48	50		{url: site + "/hello.md", status: http.StatusOK, body: "# hello"},
49	51		{url: site + "/hello.lxt", status: http.StatusOK, body: "# hello"},
50	52		{url: site + "/raw/hello", status: http.StatusOK, body: "# hello"},
	53	+	{url: site + "/raw", status: http.StatusOK, body: "<html"},
	54	+	{url: site + "/raw.md", status: http.StatusOK, body: "# raw"},
	55	+	{url: site + "/raw/raw", status: http.StatusOK, body: "# raw"},
51	56		{url: site + "/missing", status: http.StatusNotFound},
52	57		// Images are served from the user's bucket, which they don't have.
53	58		{url: site + "/cat.png", status: http.StatusUnprocessableEntity, body: "bucket does not exist"},

Commit def8a10

Commit `def8a10`