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