Antonio Mika
·
2026-10-25
1package pipe
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "net/http"
11 "net/url"
12 "regexp"
13 "strings"
14 "sync"
15 "time"
16
17 "github.com/gorilla/websocket"
18 "github.com/picosh/pico/pkg/db/postgres"
19 "github.com/picosh/pico/pkg/metrics"
20 "github.com/picosh/pico/pkg/shared"
21 "github.com/picosh/pico/pkg/shared/router"
22 "github.com/picosh/utils/pipe"
23 "github.com/prometheus/client_golang/prometheus"
24 "github.com/prometheus/client_golang/prometheus/promhttp"
25 "golang.org/x/crypto/ssh"
26)
27
28var (
29 cleanRegex = regexp.MustCompile(`[^0-9a-zA-Z,/]`)
30 upgrader = websocket.Upgrader{
31 CheckOrigin: func(r *http.Request) bool {
32 return true
33 },
34 }
35)
36
37type writeFlusher struct {
38 responseWriter http.ResponseWriter
39 controller *http.ResponseController
40}
41
42func (w writeFlusher) Write(p []byte) (n int, err error) {
43 n, err = w.responseWriter.Write(p)
44 if err == nil {
45 err = w.controller.Flush()
46 }
47 return
48}
49
50var _ io.Writer = writeFlusher{}
51
52const (
53 // pongWait is how long a websocket client has to answer a ping.
54 pongWait = 60 * time.Second
55 pingPeriod = pongWait / 2
56 writeWait = 10 * time.Second
57)
58
59func (web *webRouter) handleSub(pubsub bool) http.HandlerFunc {
60 return func(w http.ResponseWriter, r *http.Request) {
61 logger := web.cfg.Logger
62
63 clientInfo := shared.NewPicoPipeClient()
64 topic, _ := url.PathUnescape(r.PathValue("topic"))
65
66 topic = cleanRegex.ReplaceAllString(topic, "")
67
68 logger.Info("sub", "topic", topic, "info", clientInfo, "pubsub", pubsub)
69
70 params := "-p"
71 if r.URL.Query().Get("persist") == "true" {
72 params += " -k"
73 }
74
75 if accessList := r.URL.Query().Get("access"); accessList != "" {
76 logger.Info("adding access list", "topic", topic, "info", clientInfo, "access", accessList)
77 cleanList := cleanRegex.ReplaceAllString(accessList, "")
78 params += fmt.Sprintf(" -a=%s", cleanList)
79 }
80
81 p, err := web.conn.run(r.Context(), fmt.Sprintf("sub %s %s", params, topic))
82 if err != nil {
83 logger.Error("sub error", "topic", topic, "info", clientInfo, "err", err.Error())
84 http.Error(w, "server error", http.StatusInternalServerError)
85 return
86 }
87 defer func() { _ = p.Close() }()
88 // A client that leaves while nothing is published ends the sub.
89 defer context.AfterFunc(r.Context(), func() { _ = p.Close() })()
90
91 if mime := r.URL.Query().Get("mime"); mime != "" {
92 w.Header().Add("Content-Type", r.URL.Query().Get("mime"))
93 }
94
95 w.WriteHeader(http.StatusOK)
96
97 _, err = io.Copy(writeFlusher{w, http.NewResponseController(w)}, p)
98 if err != nil {
99 logger.Error("sub copy error", "topic", topic, "info", clientInfo, "err", err.Error())
100 return
101 }
102 }
103}
104
105func (web *webRouter) handlePub(pubsub bool) http.HandlerFunc {
106 return func(w http.ResponseWriter, r *http.Request) {
107 logger := web.cfg.Logger
108
109 clientInfo := shared.NewPicoPipeClient()
110 topic, _ := url.PathUnescape(r.PathValue("topic"))
111
112 topic = cleanRegex.ReplaceAllString(topic, "")
113
114 logger.Info("pub", "topic", topic, "info", clientInfo)
115
116 params := "-p"
117 if pubsub {
118 params += " -b=false"
119 }
120
121 if accessList := r.URL.Query().Get("access"); accessList != "" {
122 logger.Info("adding access list", "topic", topic, "info", clientInfo, "access", accessList)
123 cleanList := cleanRegex.ReplaceAllString(accessList, "")
124 params += fmt.Sprintf(" -a=%s", cleanList)
125 }
126
127 prefix := r.URL.Query().Get("prefix")
128
129 var wg sync.WaitGroup
130
131 reader := bufio.NewReaderSize(r.Body, 1)
132
133 first := make([]byte, 1)
134
135 nFirst, err := reader.Read(first)
136 if err != nil && !errors.Is(err, io.EOF) {
137 logger.Error("pub peek error", "topic", topic, "info", clientInfo, "err", err.Error())
138 http.Error(w, "server error", http.StatusInternalServerError)
139 return
140 }
141
142 if nFirst == 0 {
143 params += " -e"
144 }
145
146 p, err := web.conn.run(r.Context(), fmt.Sprintf("pub %s %s", params, topic))
147 if err != nil {
148 logger.Error("pub error", "topic", topic, "info", clientInfo, "err", err.Error())
149 http.Error(w, "server error", http.StatusInternalServerError)
150 return
151 }
152 defer func() { _ = p.Close() }()
153 defer context.AfterFunc(r.Context(), func() { _ = p.Close() })()
154
155 var scanErr error
156 scanStatus := http.StatusInternalServerError
157
158 wg.Add(1)
159
160 go func() {
161 defer wg.Done()
162
163 s := bufio.NewScanner(p)
164 s.Buffer(make([]byte, 32*1024), 32*1024)
165
166 for s.Scan() {
167 if s.Text() == "sending msg ..." {
168 time.Sleep(10 * time.Millisecond)
169 break
170 }
171
172 if strings.HasPrefix(s.Text(), " ssh ") {
173 f := strings.Fields(s.Text())
174 if len(f) > 1 && f[len(f)-1] != topic {
175 scanErr = fmt.Errorf("pub is not same as used, expected `%s` and received `%s`", topic, f[len(f)-1])
176 scanStatus = http.StatusUnauthorized
177 return
178 }
179 }
180 }
181
182 if err := s.Err(); err != nil {
183 scanErr = err
184 return
185 }
186 }()
187
188 wg.Wait()
189
190 if scanErr != nil {
191 logger.Error("pub scan error", "topic", topic, "info", clientInfo, "err", scanErr.Error())
192
193 msg := "server error"
194 if scanStatus == http.StatusUnauthorized {
195 msg = "access denied"
196 }
197
198 http.Error(w, msg, scanStatus)
199 return
200 }
201
202 messageToWrite := first
203 if prefix != "" {
204 messageToWrite = append([]byte(prefix), messageToWrite...)
205 }
206
207 _, err = p.Write(messageToWrite)
208 if err != nil {
209 logger.Error("pub write error", "topic", topic, "info", clientInfo, "err", err.Error())
210 http.Error(w, "server error", http.StatusInternalServerError)
211 return
212 }
213
214 _, err = io.Copy(p, reader)
215 if err != nil {
216 logger.Error("pub copy error", "topic", topic, "info", clientInfo, "err", err.Error())
217 http.Error(w, "server error", http.StatusInternalServerError)
218 return
219 }
220
221 // pub sends the message once its stdin ends, and exits when it has.
222 // Older pipe servers close the session without an exit status.
223 _ = p.CloseWrite()
224 var missing *ssh.ExitMissingError
225 if err := p.Wait(); err != nil && !errors.As(err, &missing) {
226 logger.Error("pub exit error", "topic", topic, "info", clientInfo, "err", err.Error())
227 http.Error(w, "server error", http.StatusInternalServerError)
228 return
229 }
230
231 w.WriteHeader(http.StatusOK)
232 }
233}
234
235func (web *webRouter) handlePipe() http.HandlerFunc {
236 return func(w http.ResponseWriter, r *http.Request) {
237 logger := web.cfg.Logger
238
239 c, err := upgrader.Upgrade(w, r, nil)
240 if err != nil {
241 logger.Error("pipe upgrade error", "err", err.Error())
242 return
243 }
244
245 defer func() {
246 _ = c.Close()
247 }()
248
249 clientInfo := shared.NewPicoPipeClient()
250 topic, _ := url.PathUnescape(r.PathValue("topic"))
251
252 topic = cleanRegex.ReplaceAllString(topic, "")
253
254 logger.Info("pipe", "topic", topic, "info", clientInfo)
255
256 params := "-p -c"
257 if r.URL.Query().Get("status") == "true" {
258 params = params[:len(params)-3]
259 }
260
261 if r.URL.Query().Get("replay") == "true" {
262 params += " -r"
263 }
264
265 messageType := websocket.TextMessage
266 if r.URL.Query().Get("binary") == "true" {
267 messageType = websocket.BinaryMessage
268 }
269
270 if accessList := r.URL.Query().Get("access"); accessList != "" {
271 logger.Info("adding access list", "topic", topic, "info", clientInfo, "access", accessList)
272 cleanList := cleanRegex.ReplaceAllString(accessList, "")
273 params += fmt.Sprintf(" -a=%s", cleanList)
274 }
275
276 prefix := r.URL.Query().Get("prefix")
277
278 p, err := web.conn.run(r.Context(), fmt.Sprintf("pipe %s %s", params, topic))
279 if err != nil {
280 logger.Error("pipe error", "topic", topic, "info", clientInfo, "err", err.Error())
281 _ = c.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "server error"), time.Now().Add(time.Second))
282 return
283 }
284 defer func() { _ = p.Close() }()
285
286 // A client that stops answering pings is gone, even if its
287 // connection never closed.
288 _ = c.SetReadDeadline(time.Now().Add(pongWait))
289 c.SetPongHandler(func(string) error {
290 return c.SetReadDeadline(time.Now().Add(pongWait))
291 })
292 done := make(chan struct{})
293 defer close(done)
294 go func() {
295 ticker := time.NewTicker(pingPeriod)
296 defer ticker.Stop()
297 for {
298 select {
299 case <-done:
300 return
301 case <-ticker.C:
302 if err := c.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeWait)); err != nil {
303 _ = c.Close()
304 return
305 }
306 }
307 }
308 }()
309
310 var wg sync.WaitGroup
311 wg.Add(2)
312
313 go func() {
314 defer func() {
315 _ = p.Close()
316 _ = c.Close()
317 wg.Done()
318 }()
319
320 for {
321 _, message, err := c.ReadMessage()
322 if err != nil {
323 logger.Error("pipe read error", "topic", topic, "info", clientInfo, "err", err.Error())
324 break
325 }
326
327 _, err = p.Write(message)
328 if err != nil {
329 logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error())
330 break
331 }
332 }
333 }()
334
335 go func() {
336 defer func() {
337 _ = p.Close()
338 _ = c.Close()
339 wg.Done()
340 }()
341
342 var messageBuffer []byte
343 buf := make([]byte, 32*1024)
344
345 for {
346 n, err := p.Read(buf)
347 if err != nil {
348 logger.Error("pipe read error", "topic", topic, "info", clientInfo, "err", err.Error())
349 break
350 }
351
352 messageBuffer = append(messageBuffer, buf[:n]...)
353
354 if prefix != "" {
355 // Buffer and split on prefix boundaries
356 for {
357 firstIdx := bytes.Index(messageBuffer, []byte(prefix))
358 if firstIdx == -1 {
359 // No prefix found, clear buffer (shouldn't happen in normal use)
360 messageBuffer = nil
361 break
362 }
363
364 // Look for next prefix after the first one
365 secondIdx := bytes.Index(messageBuffer[firstIdx+len(prefix):], []byte(prefix))
366 if secondIdx == -1 {
367 // No complete message yet, keep buffer as is
368 break
369 }
370
371 // We have a complete message, extract and send it
372 messageToSend := messageBuffer[firstIdx : firstIdx+len(prefix)+secondIdx]
373 err = c.WriteMessage(messageType, messageToSend)
374 if err != nil {
375 logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error())
376 break
377 }
378
379 // Update buffer to remove sent message
380 messageBuffer = messageBuffer[firstIdx+len(prefix)+secondIdx:]
381 }
382 } else {
383 // No prefix set, send all data as-is
384 if len(messageBuffer) > 0 {
385 err = c.WriteMessage(messageType, messageBuffer)
386 if err != nil {
387 logger.Error("pipe write error", "topic", topic, "info", clientInfo, "err", err.Error())
388 break
389 }
390 messageBuffer = nil
391 }
392 }
393 }
394 }()
395
396 wg.Wait()
397 }
398}
399
400func rssHandler(cfg *Config, dbpool DB) http.HandlerFunc {
401 return func(w http.ResponseWriter, r *http.Request) {
402 apiToken, _ := url.PathUnescape(r.PathValue("token"))
403 user, err := dbpool.FindUserByToken(apiToken)
404 if err != nil {
405 cfg.Logger.Error(
406 "could not find user for token",
407 "err", err.Error(),
408 "token", apiToken,
409 )
410 http.Error(w, "invalid token", http.StatusNotFound)
411 return
412 }
413 router.SetUser(r, user.Name)
414 rss, err := MonitorRss(dbpool, user, cfg.Domain)
415 if err != nil {
416 cfg.Logger.Error(
417 "error generating monitor rss feed",
418 "err", err,
419 "token", apiToken,
420 )
421 http.Error(w, "error generating monitor rss feed", http.StatusInternalServerError)
422 return
423 }
424
425 _, err = w.Write([]byte(rss))
426 if err != nil {
427 cfg.Logger.Error(
428 "error with rss response writer",
429 "err", err,
430 "token", apiToken,
431 )
432 http.Error(w, "error generating monitor rss feederror with rss response writer", http.StatusInternalServerError)
433 return
434 }
435 }
436}
437
438type webRouter struct {
439 cfg *Config
440 db DB
441 conn *pipeConn
442}
443
444func (web *webRouter) static(mux *http.ServeMux) {
445 cfg := web.cfg
446 files := cfg.Static()
447 mux.HandleFunc("GET /favicon-16x16.png", files.File("favicon-16x16.png", "image/png"))
448 mux.HandleFunc("GET /apple-touch-icon.png", files.File("apple-touch-icon.png", "image/png"))
449 mux.HandleFunc("GET /favicon.ico", files.File("favicon.ico", "image/x-icon"))
450 mux.HandleFunc("GET /robots.txt", files.File("robots.txt", "text/plain"))
451 mux.HandleFunc("GET /anim.js", files.File("anim.js", "text/javascript"))
452 if cfg.Debug {
453 router.CreatePProfRoutesMux(mux)
454 }
455}
456
457func (web *webRouter) mainMux() *http.ServeMux {
458 mux := http.NewServeMux()
459 mux.HandleFunc("GET /{$}", web.cfg.Static().Page("html/marketing.page.tmpl"))
460 mux.HandleFunc("GET /check", router.CheckHandler(web.cfg.Domain, web.cfg.Space, web.db))
461 mux.HandleFunc("GET /rss/{token...}", rssHandler(web.cfg, web.db))
462 mux.Handle("GET /_metrics", promhttp.Handler())
463
464 mux.HandleFunc("GET /topic/{topic...}", router.Cors(web.handleSub(false)))
465 mux.HandleFunc("POST /topic/{topic...}", router.Cors(web.handlePub(false)))
466 mux.HandleFunc("OPTIONS /topic/{topic...}", router.CorsPreflight)
467 mux.HandleFunc("GET /pubsub/{topic...}", router.Cors(web.handleSub(true)))
468 mux.HandleFunc("POST /pubsub/{topic...}", router.Cors(web.handlePub(true)))
469 mux.HandleFunc("OPTIONS /pubsub/{topic...}", router.CorsPreflight)
470 mux.HandleFunc("GET /pipe/{topic...}", router.Cors(web.handlePipe()))
471 mux.HandleFunc("OPTIONS /pipe/{topic...}", router.CorsPreflight)
472
473 web.static(mux)
474 return mux
475}
476
477// sitesMux serves only static files on subdomains.
478func (web *webRouter) sitesMux() *http.ServeMux {
479 mux := http.NewServeMux()
480 web.static(mux)
481 return mux
482}
483
484func StartApiServer() {
485 cfg := NewConfig("pipe-web")
486 db := NewDB(postgres.NewDB(cfg.DbURL, cfg.Logger))
487 defer func() {
488 _ = db.Close()
489 }()
490 logger := cfg.Logger
491
492 info := shared.NewPicoPipeClient()
493
494 pingClient, err := pipe.NewClient(context.Background(), logger.With("info", info), info)
495 if err != nil {
496 panic(err)
497 }
498
499 pingSession, err := pingClient.AddSession("ping", "pub -b=false -c ping", 0, -1, -1)
500 if err != nil {
501 panic(err)
502 }
503
504 go func() {
505 for {
506 _, err := fmt.Fprintf(pingSession, "%s: pipe-web ping\n", time.Now().UTC().Format(time.RFC3339))
507 if err != nil {
508 logger.Error("pipe ping error", "err", err.Error())
509 }
510
511 time.Sleep(5 * time.Second)
512 }
513 }()
514
515 web := &webRouter{cfg: cfg, db: db, conn: newPipeConn(info)}
516 handler := router.NewHostRouter(cfg.Domain, cfg.Space, web.mainMux(), web.sitesMux())
517 handler = router.NewUserMetrics("pipe", prometheus.DefaultRegisterer, router.UserMetricsOpts{}).Handler(handler)
518 metrics.RegisterServerInfo(prometheus.DefaultRegisterer, "pipe")
519
520 portStr := fmt.Sprintf(":%s", cfg.Port)
521 logger.Info(
522 "Starting server on port",
523 "port", cfg.Port,
524 "domain", cfg.Domain,
525 )
526
527 logger.Error("listen", "err", http.ListenAndServe(portStr, handler).Error())
528}