server.go 39 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261
  1. package server
  2. import (
  3. "bytes"
  4. "context"
  5. "embed"
  6. "encoding/base64"
  7. "encoding/json"
  8. "errors"
  9. "fmt"
  10. "github.com/emersion/go-smtp"
  11. "github.com/gorilla/websocket"
  12. "golang.org/x/sync/errgroup"
  13. "heckel.io/ntfy/auth"
  14. "heckel.io/ntfy/util"
  15. "io"
  16. "log"
  17. "net"
  18. "net/http"
  19. "net/http/httptest"
  20. "net/url"
  21. "os"
  22. "path"
  23. "path/filepath"
  24. "regexp"
  25. "strconv"
  26. "strings"
  27. "sync"
  28. "time"
  29. "unicode/utf8"
  30. )
  31. // Server is the main server, providing the UI and API for ntfy
  32. type Server struct {
  33. config *Config
  34. httpServer *http.Server
  35. httpsServer *http.Server
  36. unixListener net.Listener
  37. smtpServer *smtp.Server
  38. smtpBackend *smtpBackend
  39. topics map[string]*topic
  40. visitors map[string]*visitor
  41. firebase subscriber
  42. mailer mailer
  43. messages int64
  44. auth auth.Auther
  45. messageCache *messageCache
  46. fileCache *fileCache
  47. closeChan chan bool
  48. mu sync.Mutex
  49. }
  50. // handleFunc extends the normal http.HandlerFunc to be able to easily return errors
  51. type handleFunc func(http.ResponseWriter, *http.Request, *visitor) error
  52. var (
  53. // If changed, don't forget to update Android App and auth_sqlite.go
  54. topicRegex = regexp.MustCompile(`^[-_A-Za-z0-9]{1,64}$`) // No /!
  55. topicPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}$`) // Regex must match JS & Android app!
  56. updateTopicPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}/[A-Za-z0-9]{12}$`) // ID length must match messageIDLength & util.randomStringCharset
  57. externalTopicPathRegex = regexp.MustCompile(`^/[^/]+\.[^/]+/[-_A-Za-z0-9]{1,64}$`) // Extended topic path, for web-app, e.g. /example.com/mytopic
  58. jsonPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/json$`)
  59. ssePathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/sse$`)
  60. rawPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/raw$`)
  61. wsPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/ws$`)
  62. authPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/auth$`)
  63. publishPathRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}/(publish|send|trigger)$`)
  64. webConfigPath = "/config.js"
  65. staticRegex = regexp.MustCompile(`^/static/.+`)
  66. docsRegex = regexp.MustCompile(`^/docs(|/.*)$`)
  67. fileRegex = regexp.MustCompile(`^/file/([-_A-Za-z0-9]{1,64})(?:\.[A-Za-z0-9]{1,16})?$`)
  68. disallowedTopics = []string{"docs", "static", "file", "app", "settings"} // If updated, also update in Android app
  69. attachURLRegex = regexp.MustCompile(`^https?://`)
  70. //go:embed "example.html"
  71. exampleSource string
  72. //go:embed site
  73. webFs embed.FS
  74. webFsCached = &util.CachingEmbedFS{ModTime: time.Now(), FS: webFs}
  75. webSiteDir = "/site"
  76. webHomeIndex = "/home.html" // Landing page, only if "web-root: home"
  77. webAppIndex = "/app.html" // React app
  78. //go:embed docs
  79. docsStaticFs embed.FS
  80. docsStaticCached = &util.CachingEmbedFS{ModTime: time.Now(), FS: docsStaticFs}
  81. )
  82. const (
  83. firebaseControlTopic = "~control" // See Android if changed
  84. emptyMessageBody = "triggered" // Used if message body is empty
  85. defaultAttachmentMessage = "You received a file: %s" // Used if message body is empty, and there is an attachment
  86. encodingBase64 = "base64"
  87. )
  88. // WebSocket constants
  89. const (
  90. wsWriteWait = 2 * time.Second
  91. wsBufferSize = 1024
  92. wsReadLimit = 64 // We only ever receive PINGs
  93. wsPongWait = 15 * time.Second
  94. )
  95. // New instantiates a new Server. It creates the cache and adds a Firebase
  96. // subscriber (if configured).
  97. func New(conf *Config) (*Server, error) {
  98. var mailer mailer
  99. if conf.SMTPSenderAddr != "" {
  100. mailer = &smtpSender{config: conf}
  101. }
  102. messageCache, err := createMessageCache(conf)
  103. if err != nil {
  104. return nil, err
  105. }
  106. topics, err := messageCache.Topics()
  107. if err != nil {
  108. return nil, err
  109. }
  110. var fileCache *fileCache
  111. if conf.AttachmentCacheDir != "" {
  112. fileCache, err = newFileCache(conf.AttachmentCacheDir, conf.AttachmentTotalSizeLimit, conf.AttachmentFileSizeLimit)
  113. if err != nil {
  114. return nil, err
  115. }
  116. }
  117. var auther auth.Auther
  118. if conf.AuthFile != "" {
  119. auther, err = auth.NewSQLiteAuth(conf.AuthFile, conf.AuthDefaultRead, conf.AuthDefaultWrite)
  120. if err != nil {
  121. return nil, err
  122. }
  123. }
  124. var firebaseSubscriber subscriber
  125. if conf.FirebaseKeyFile != "" {
  126. var err error
  127. firebaseSubscriber, err = createFirebaseSubscriber(conf.FirebaseKeyFile, auther)
  128. if err != nil {
  129. return nil, err
  130. }
  131. }
  132. return &Server{
  133. config: conf,
  134. messageCache: messageCache,
  135. fileCache: fileCache,
  136. firebase: firebaseSubscriber,
  137. mailer: mailer,
  138. topics: topics,
  139. auth: auther,
  140. visitors: make(map[string]*visitor),
  141. }, nil
  142. }
  143. func createMessageCache(conf *Config) (*messageCache, error) {
  144. if conf.CacheDuration == 0 {
  145. return newNopCache()
  146. } else if conf.CacheFile != "" {
  147. return newSqliteCache(conf.CacheFile, false)
  148. }
  149. return newMemCache()
  150. }
  151. // Run executes the main server. It listens on HTTP (+ HTTPS, if configured), and starts
  152. // a manager go routine to print stats and prune messages.
  153. func (s *Server) Run() error {
  154. var listenStr string
  155. if s.config.ListenHTTP != "" {
  156. listenStr += fmt.Sprintf(" %s[http]", s.config.ListenHTTP)
  157. }
  158. if s.config.ListenHTTPS != "" {
  159. listenStr += fmt.Sprintf(" %s[https]", s.config.ListenHTTPS)
  160. }
  161. if s.config.ListenUnix != "" {
  162. listenStr += fmt.Sprintf(" %s[unix]", s.config.ListenUnix)
  163. }
  164. if s.config.SMTPServerListen != "" {
  165. listenStr += fmt.Sprintf(" %s[smtp]", s.config.SMTPServerListen)
  166. }
  167. log.Printf("Listening on%s", listenStr)
  168. mux := http.NewServeMux()
  169. mux.HandleFunc("/", s.handle)
  170. errChan := make(chan error)
  171. s.mu.Lock()
  172. s.closeChan = make(chan bool)
  173. if s.config.ListenHTTP != "" {
  174. s.httpServer = &http.Server{Addr: s.config.ListenHTTP, Handler: mux}
  175. go func() {
  176. errChan <- s.httpServer.ListenAndServe()
  177. }()
  178. }
  179. if s.config.ListenHTTPS != "" {
  180. s.httpsServer = &http.Server{Addr: s.config.ListenHTTPS, Handler: mux}
  181. go func() {
  182. errChan <- s.httpsServer.ListenAndServeTLS(s.config.CertFile, s.config.KeyFile)
  183. }()
  184. }
  185. if s.config.ListenUnix != "" {
  186. go func() {
  187. var err error
  188. s.mu.Lock()
  189. os.Remove(s.config.ListenUnix)
  190. s.unixListener, err = net.Listen("unix", s.config.ListenUnix)
  191. if err != nil {
  192. errChan <- err
  193. return
  194. }
  195. s.mu.Unlock()
  196. httpServer := &http.Server{Handler: mux}
  197. errChan <- httpServer.Serve(s.unixListener)
  198. }()
  199. }
  200. if s.config.SMTPServerListen != "" {
  201. go func() {
  202. errChan <- s.runSMTPServer()
  203. }()
  204. }
  205. s.mu.Unlock()
  206. go s.runManager()
  207. go s.runAtSender()
  208. go s.runFirebaseKeepaliver()
  209. return <-errChan
  210. }
  211. // Stop stops HTTP (+HTTPS) server and all managers
  212. func (s *Server) Stop() {
  213. s.mu.Lock()
  214. defer s.mu.Unlock()
  215. if s.httpServer != nil {
  216. s.httpServer.Close()
  217. }
  218. if s.httpsServer != nil {
  219. s.httpsServer.Close()
  220. }
  221. if s.unixListener != nil {
  222. s.unixListener.Close()
  223. }
  224. if s.smtpServer != nil {
  225. s.smtpServer.Close()
  226. }
  227. close(s.closeChan)
  228. }
  229. func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
  230. v := s.visitor(r)
  231. if err := s.handleInternal(w, r, v); err != nil {
  232. if websocket.IsWebSocketUpgrade(r) {
  233. log.Printf("[%s] WS %s %s - %s", v.ip, r.Method, r.URL.Path, err.Error())
  234. return // Do not attempt to write to upgraded connection
  235. }
  236. httpErr, ok := err.(*errHTTP)
  237. if !ok {
  238. httpErr = errHTTPInternalError
  239. }
  240. log.Printf("[%s] HTTP %s %s - %d - %d - %s", v.ip, r.Method, r.URL.Path, httpErr.HTTPCode, httpErr.Code, err.Error())
  241. w.Header().Set("Content-Type", "application/json")
  242. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  243. w.WriteHeader(httpErr.HTTPCode)
  244. io.WriteString(w, httpErr.JSON()+"\n")
  245. }
  246. }
  247. func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visitor) error {
  248. if r.Method == http.MethodGet && r.URL.Path == "/" {
  249. return s.handleHome(w, r)
  250. } else if r.Method == http.MethodGet && r.URL.Path == "/example.html" {
  251. return s.handleExample(w, r)
  252. } else if r.Method == http.MethodHead && r.URL.Path == "/" {
  253. return s.handleEmpty(w, r, v)
  254. } else if r.Method == http.MethodGet && r.URL.Path == webConfigPath {
  255. return s.handleWebConfig(w, r)
  256. } else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
  257. return s.handleStatic(w, r)
  258. } else if r.Method == http.MethodGet && docsRegex.MatchString(r.URL.Path) {
  259. return s.handleDocs(w, r)
  260. } else if r.Method == http.MethodGet && fileRegex.MatchString(r.URL.Path) && s.config.AttachmentCacheDir != "" {
  261. return s.limitRequests(s.handleFile)(w, r, v)
  262. } else if r.Method == http.MethodOptions {
  263. return s.handleOptions(w, r)
  264. } else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && r.URL.Path == "/" {
  265. return s.limitRequests(s.transformBodyJSON(s.authWrite(s.handlePublish)))(w, r, v)
  266. } else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && (topicPathRegex.MatchString(r.URL.Path) || updateTopicPathRegex.MatchString(r.URL.Path)) {
  267. return s.limitRequests(s.authWrite(s.handlePublish))(w, r, v)
  268. } else if r.Method == http.MethodGet && publishPathRegex.MatchString(r.URL.Path) {
  269. return s.limitRequests(s.authWrite(s.handlePublish))(w, r, v)
  270. } else if r.Method == http.MethodGet && jsonPathRegex.MatchString(r.URL.Path) {
  271. return s.limitRequests(s.authRead(s.handleSubscribeJSON))(w, r, v)
  272. } else if r.Method == http.MethodGet && ssePathRegex.MatchString(r.URL.Path) {
  273. return s.limitRequests(s.authRead(s.handleSubscribeSSE))(w, r, v)
  274. } else if r.Method == http.MethodGet && rawPathRegex.MatchString(r.URL.Path) {
  275. return s.limitRequests(s.authRead(s.handleSubscribeRaw))(w, r, v)
  276. } else if r.Method == http.MethodGet && wsPathRegex.MatchString(r.URL.Path) {
  277. return s.limitRequests(s.authRead(s.handleSubscribeWS))(w, r, v)
  278. } else if r.Method == http.MethodGet && authPathRegex.MatchString(r.URL.Path) {
  279. return s.limitRequests(s.authRead(s.handleTopicAuth))(w, r, v)
  280. } else if r.Method == http.MethodGet && (topicPathRegex.MatchString(r.URL.Path) || externalTopicPathRegex.MatchString(r.URL.Path)) {
  281. return s.handleTopic(w, r)
  282. }
  283. return errHTTPNotFound
  284. }
  285. func (s *Server) handleHome(w http.ResponseWriter, r *http.Request) error {
  286. if s.config.WebRootIsApp {
  287. r.URL.Path = webAppIndex
  288. } else {
  289. r.URL.Path = webHomeIndex
  290. }
  291. return s.handleStatic(w, r)
  292. }
  293. func (s *Server) handleTopic(w http.ResponseWriter, r *http.Request) error {
  294. unifiedpush := readBoolParam(r, false, "x-unifiedpush", "unifiedpush", "up") // see PUT/POST too!
  295. if unifiedpush {
  296. w.Header().Set("Content-Type", "application/json")
  297. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  298. _, err := io.WriteString(w, `{"unifiedpush":{"version":1}}`+"\n")
  299. return err
  300. }
  301. r.URL.Path = webAppIndex
  302. return s.handleStatic(w, r)
  303. }
  304. func (s *Server) handleEmpty(_ http.ResponseWriter, _ *http.Request, _ *visitor) error {
  305. return nil
  306. }
  307. func (s *Server) handleTopicAuth(w http.ResponseWriter, _ *http.Request, _ *visitor) error {
  308. w.Header().Set("Content-Type", "application/json")
  309. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  310. _, err := io.WriteString(w, `{"success":true}`+"\n")
  311. return err
  312. }
  313. func (s *Server) handleExample(w http.ResponseWriter, _ *http.Request) error {
  314. _, err := io.WriteString(w, exampleSource)
  315. return err
  316. }
  317. func (s *Server) handleWebConfig(w http.ResponseWriter, r *http.Request) error {
  318. appRoot := "/"
  319. if !s.config.WebRootIsApp {
  320. appRoot = "/app"
  321. }
  322. disallowedTopicsStr := `"` + strings.Join(disallowedTopics, `", "`) + `"`
  323. w.Header().Set("Content-Type", "text/javascript")
  324. _, err := io.WriteString(w, fmt.Sprintf(`// Generated server configuration
  325. var config = {
  326. appRoot: "%s",
  327. disallowedTopics: [%s]
  328. };`, appRoot, disallowedTopicsStr))
  329. return err
  330. }
  331. func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
  332. r.URL.Path = webSiteDir + r.URL.Path
  333. util.Gzip(http.FileServer(http.FS(webFsCached))).ServeHTTP(w, r)
  334. return nil
  335. }
  336. func (s *Server) handleDocs(w http.ResponseWriter, r *http.Request) error {
  337. util.Gzip(http.FileServer(http.FS(docsStaticCached))).ServeHTTP(w, r)
  338. return nil
  339. }
  340. func (s *Server) handleFile(w http.ResponseWriter, r *http.Request, v *visitor) error {
  341. if s.config.AttachmentCacheDir == "" {
  342. return errHTTPInternalError
  343. }
  344. matches := fileRegex.FindStringSubmatch(r.URL.Path)
  345. if len(matches) != 2 {
  346. return errHTTPInternalErrorInvalidFilePath
  347. }
  348. messageID := matches[1]
  349. file := filepath.Join(s.config.AttachmentCacheDir, messageID)
  350. stat, err := os.Stat(file)
  351. if err != nil {
  352. return errHTTPNotFound
  353. }
  354. if err := v.BandwidthLimiter().Allow(stat.Size()); err != nil {
  355. return errHTTPTooManyRequestsAttachmentBandwidthLimit
  356. }
  357. w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size()))
  358. f, err := os.Open(file)
  359. if err != nil {
  360. return err
  361. }
  362. defer f.Close()
  363. _, err = io.Copy(util.NewContentTypeWriter(w, r.URL.Path), f)
  364. return err
  365. }
  366. func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
  367. t, messageID, err := s.topicAndMessageIDFromPath(r.URL.Path)
  368. if err != nil {
  369. return err
  370. }
  371. var m *message
  372. update := messageID != ""
  373. if update {
  374. m, err = s.messageCache.Message(t.ID, messageID)
  375. if err != nil {
  376. return errHTTPNotFoundMessageID
  377. }
  378. newUpdated := time.Now().Unix()
  379. if newUpdated <= m.Updated {
  380. return errHTTPTooManyRequestsUpdatingTooQuickly
  381. }
  382. m.Updated = newUpdated
  383. } else {
  384. m = newDefaultMessage(t.ID, "")
  385. }
  386. cache, firebase, email, unifiedpush, err := s.parsePublishParams(r, v, m)
  387. if err != nil {
  388. return err
  389. }
  390. body, err := util.Peak(r.Body, s.config.MessageLimit)
  391. if err != nil {
  392. return err
  393. }
  394. if err := s.handlePublishBody(r, v, m, body, unifiedpush); err != nil {
  395. return err
  396. }
  397. if m.Message == "" {
  398. m.Message = emptyMessageBody
  399. }
  400. delayed := m.Time > time.Now().Unix()
  401. if !delayed {
  402. if err := t.Publish(m); err != nil {
  403. return err
  404. }
  405. }
  406. if s.firebase != nil && firebase && !delayed {
  407. go func() {
  408. if err := s.firebase(m); err != nil {
  409. log.Printf("[%s] FB - Unable to publish to Firebase: %v", v.ip, err.Error())
  410. }
  411. }()
  412. }
  413. if s.mailer != nil && email != "" && !delayed {
  414. go func() {
  415. if err := s.mailer.Send(v.ip, email, m); err != nil {
  416. log.Printf("[%s] MAIL - Unable to send email: %v", v.ip, err.Error())
  417. }
  418. }()
  419. }
  420. if cache {
  421. if update {
  422. if err := s.messageCache.UpdateMessage(m); err != nil {
  423. return err
  424. }
  425. } else {
  426. if err := s.messageCache.AddMessage(m); err != nil {
  427. return err
  428. }
  429. }
  430. }
  431. w.Header().Set("Content-Type", "application/json")
  432. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  433. if err := json.NewEncoder(w).Encode(m); err != nil {
  434. return err
  435. }
  436. s.mu.Lock()
  437. s.messages++
  438. s.mu.Unlock()
  439. return nil
  440. }
  441. func (s *Server) parsePublishParams(r *http.Request, v *visitor, m *message) (cache bool, firebase bool, email string, unifiedpush bool, err error) {
  442. cache = readBoolParam(r, true, "x-cache", "cache")
  443. if !cache && m.Updated != 0 {
  444. return false, false, "", false, errors.New("message updates must be cached")
  445. }
  446. // TODO more restrictions
  447. firebase = readBoolParam(r, true, "x-firebase", "firebase")
  448. title := readParam(r, "x-title", "title", "t")
  449. if title != "" {
  450. m.Title = title
  451. }
  452. click := readParam(r, "x-click", "click")
  453. if click != "" {
  454. m.Click = click
  455. }
  456. filename := readParam(r, "x-filename", "filename", "file", "f")
  457. attach := readParam(r, "x-attach", "attach", "a")
  458. if attach != "" || filename != "" {
  459. m.Attachment = &attachment{}
  460. }
  461. if filename != "" {
  462. m.Attachment.Name = filename
  463. }
  464. if attach != "" {
  465. if !attachURLRegex.MatchString(attach) {
  466. return false, false, "", false, errHTTPBadRequestAttachmentURLInvalid
  467. }
  468. m.Attachment.URL = attach
  469. if m.Attachment.Name == "" {
  470. u, err := url.Parse(m.Attachment.URL)
  471. if err == nil {
  472. m.Attachment.Name = path.Base(u.Path)
  473. if m.Attachment.Name == "." || m.Attachment.Name == "/" {
  474. m.Attachment.Name = ""
  475. }
  476. }
  477. }
  478. if m.Attachment.Name == "" {
  479. m.Attachment.Name = "attachment"
  480. }
  481. }
  482. email = readParam(r, "x-email", "x-e-mail", "email", "e-mail", "mail", "e")
  483. if email != "" {
  484. if err := v.EmailAllowed(); err != nil {
  485. return false, false, "", false, errHTTPTooManyRequestsLimitEmails
  486. }
  487. }
  488. if s.mailer == nil && email != "" {
  489. return false, false, "", false, errHTTPBadRequestEmailDisabled
  490. }
  491. messageStr := strings.ReplaceAll(readParam(r, "x-message", "message", "m"), "\\n", "\n")
  492. if messageStr != "" {
  493. m.Message = messageStr
  494. }
  495. priority, err := util.ParsePriority(readParam(r, "x-priority", "priority", "prio", "p"))
  496. if err != nil {
  497. return false, false, "", false, errHTTPBadRequestPriorityInvalid
  498. } else if priority > 0 {
  499. m.Priority = priority
  500. }
  501. tagsStr := readParam(r, "x-tags", "tags", "tag", "ta")
  502. if tagsStr != "" {
  503. m.Tags = make([]string, 0)
  504. for _, s := range util.SplitNoEmpty(tagsStr, ",") {
  505. m.Tags = append(m.Tags, strings.TrimSpace(s))
  506. }
  507. }
  508. delayStr := readParam(r, "x-delay", "delay", "x-at", "at", "x-in", "in")
  509. if delayStr != "" {
  510. if !cache {
  511. return false, false, "", false, errHTTPBadRequestDelayNoCache
  512. }
  513. if email != "" {
  514. return false, false, "", false, errHTTPBadRequestDelayNoEmail // we cannot store the email address (yet)
  515. }
  516. delay, err := util.ParseFutureTime(delayStr, time.Now())
  517. if err != nil {
  518. return false, false, "", false, errHTTPBadRequestDelayCannotParse
  519. } else if delay.Unix() < time.Now().Add(s.config.MinDelay).Unix() {
  520. return false, false, "", false, errHTTPBadRequestDelayTooSmall
  521. } else if delay.Unix() > time.Now().Add(s.config.MaxDelay).Unix() {
  522. return false, false, "", false, errHTTPBadRequestDelayTooLarge
  523. }
  524. m.Time = delay.Unix()
  525. }
  526. unifiedpush = readBoolParam(r, false, "x-unifiedpush", "unifiedpush", "up") // see GET too!
  527. if unifiedpush {
  528. firebase = false
  529. unifiedpush = true
  530. }
  531. return cache, firebase, email, unifiedpush, nil
  532. }
  533. // handlePublishBody consumes the PUT/POST body and decides whether the body is an attachment or the message.
  534. //
  535. // 1. curl -T somebinarydata.bin "ntfy.sh/mytopic?up=1"
  536. // If body is binary, encode as base64, if not do not encode
  537. // 2. curl -H "Attach: http://example.com/file.jpg" ntfy.sh/mytopic
  538. // Body must be a message, because we attached an external URL
  539. // 3. curl -T short.txt -H "Filename: short.txt" ntfy.sh/mytopic
  540. // Body must be attachment, because we passed a filename
  541. // 4. curl -T file.txt ntfy.sh/mytopic
  542. // If file.txt is <= 4096 (message limit) and valid UTF-8, treat it as a message
  543. // 5. curl -T file.txt ntfy.sh/mytopic
  544. // If file.txt is > message limit, treat it as an attachment
  545. func (s *Server) handlePublishBody(r *http.Request, v *visitor, m *message, body *util.PeakedReadCloser, unifiedpush bool) error {
  546. if unifiedpush {
  547. return s.handleBodyAsMessageAutoDetect(m, body) // Case 1
  548. } else if m.Attachment != nil && m.Attachment.URL != "" {
  549. return s.handleBodyAsTextMessage(m, body) // Case 2
  550. } else if m.Attachment != nil && m.Attachment.Name != "" {
  551. return s.handleBodyAsAttachment(r, v, m, body) // Case 3
  552. } else if !body.LimitReached && utf8.Valid(body.PeakedBytes) {
  553. return s.handleBodyAsTextMessage(m, body) // Case 4
  554. }
  555. return s.handleBodyAsAttachment(r, v, m, body) // Case 5
  556. }
  557. func (s *Server) handleBodyAsMessageAutoDetect(m *message, body *util.PeakedReadCloser) error {
  558. if utf8.Valid(body.PeakedBytes) {
  559. m.Message = string(body.PeakedBytes) // Do not trim
  560. } else {
  561. m.Message = base64.StdEncoding.EncodeToString(body.PeakedBytes)
  562. m.Encoding = encodingBase64
  563. }
  564. return nil
  565. }
  566. func (s *Server) handleBodyAsTextMessage(m *message, body *util.PeakedReadCloser) error {
  567. if !utf8.Valid(body.PeakedBytes) {
  568. return errHTTPBadRequestMessageNotUTF8
  569. }
  570. if len(body.PeakedBytes) > 0 { // Empty body should not override message (publish via GET!)
  571. m.Message = strings.TrimSpace(string(body.PeakedBytes)) // Truncates the message to the peak limit if required
  572. }
  573. if m.Attachment != nil && m.Attachment.Name != "" && m.Message == "" {
  574. m.Message = fmt.Sprintf(defaultAttachmentMessage, m.Attachment.Name)
  575. }
  576. return nil
  577. }
  578. func (s *Server) handleBodyAsAttachment(r *http.Request, v *visitor, m *message, body *util.PeakedReadCloser) error {
  579. if s.fileCache == nil || s.config.BaseURL == "" || s.config.AttachmentCacheDir == "" {
  580. return errHTTPBadRequestAttachmentsDisallowed
  581. } else if m.Time > time.Now().Add(s.config.AttachmentExpiryDuration).Unix() {
  582. return errHTTPBadRequestAttachmentsExpiryBeforeDelivery
  583. }
  584. visitorAttachmentsSize, err := s.messageCache.AttachmentsSize(v.ip)
  585. if err != nil {
  586. return err
  587. }
  588. remainingVisitorAttachmentSize := s.config.VisitorAttachmentTotalSizeLimit - visitorAttachmentsSize
  589. contentLengthStr := r.Header.Get("Content-Length")
  590. if contentLengthStr != "" { // Early "do-not-trust" check, hard limit see below
  591. contentLength, err := strconv.ParseInt(contentLengthStr, 10, 64)
  592. if err == nil && (contentLength > remainingVisitorAttachmentSize || contentLength > s.config.AttachmentFileSizeLimit) {
  593. return errHTTPBadRequestAttachmentTooLarge
  594. }
  595. }
  596. if m.Attachment == nil {
  597. m.Attachment = &attachment{}
  598. }
  599. var ext string
  600. m.Attachment.Owner = v.ip // Important for attachment rate limiting
  601. m.Attachment.Expires = time.Now().Add(s.config.AttachmentExpiryDuration).Unix()
  602. m.Attachment.Type, ext = util.DetectContentType(body.PeakedBytes, m.Attachment.Name)
  603. m.Attachment.URL = fmt.Sprintf("%s/file/%s%s", s.config.BaseURL, m.ID, ext)
  604. if m.Attachment.Name == "" {
  605. m.Attachment.Name = fmt.Sprintf("attachment%s", ext)
  606. }
  607. if m.Message == "" {
  608. m.Message = fmt.Sprintf(defaultAttachmentMessage, m.Attachment.Name)
  609. }
  610. m.Attachment.Size, err = s.fileCache.Write(m.ID, body, v.BandwidthLimiter(), util.NewFixedLimiter(remainingVisitorAttachmentSize))
  611. if err == util.ErrLimitReached {
  612. return errHTTPBadRequestAttachmentTooLarge
  613. } else if err != nil {
  614. return err
  615. }
  616. return nil
  617. }
  618. func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request, v *visitor) error {
  619. encoder := func(msg *message) (string, error) {
  620. var buf bytes.Buffer
  621. if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
  622. return "", err
  623. }
  624. return buf.String(), nil
  625. }
  626. return s.handleSubscribeHTTP(w, r, v, "application/x-ndjson", encoder)
  627. }
  628. func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request, v *visitor) error {
  629. encoder := func(msg *message) (string, error) {
  630. var buf bytes.Buffer
  631. if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
  632. return "", err
  633. }
  634. if msg.Event != messageEvent {
  635. return fmt.Sprintf("event: %s\ndata: %s\n", msg.Event, buf.String()), nil // Browser's .onmessage() does not fire on this!
  636. }
  637. return fmt.Sprintf("data: %s\n", buf.String()), nil
  638. }
  639. return s.handleSubscribeHTTP(w, r, v, "text/event-stream", encoder)
  640. }
  641. func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request, v *visitor) error {
  642. encoder := func(msg *message) (string, error) {
  643. if msg.Event == messageEvent { // only handle default events
  644. return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
  645. }
  646. return "\n", nil // "keepalive" and "open" events just send an empty line
  647. }
  648. return s.handleSubscribeHTTP(w, r, v, "text/plain", encoder)
  649. }
  650. func (s *Server) handleSubscribeHTTP(w http.ResponseWriter, r *http.Request, v *visitor, contentType string, encoder messageEncoder) error {
  651. if err := v.SubscriptionAllowed(); err != nil {
  652. return errHTTPTooManyRequestsLimitSubscriptions
  653. }
  654. defer v.RemoveSubscription()
  655. topics, topicsStr, err := s.topicsFromPath(r.URL.Path)
  656. if err != nil {
  657. return err
  658. }
  659. poll, since, scheduled, filters, err := parseSubscribeParams(r)
  660. if err != nil {
  661. return err
  662. }
  663. var wlock sync.Mutex
  664. sub := func(msg *message) error {
  665. if !filters.Pass(msg) {
  666. return nil
  667. }
  668. m, err := encoder(msg)
  669. if err != nil {
  670. return err
  671. }
  672. wlock.Lock()
  673. defer wlock.Unlock()
  674. if _, err := w.Write([]byte(m)); err != nil {
  675. return err
  676. }
  677. if fl, ok := w.(http.Flusher); ok {
  678. fl.Flush()
  679. }
  680. return nil
  681. }
  682. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  683. w.Header().Set("Content-Type", contentType+"; charset=utf-8") // Android/Volley client needs charset!
  684. if poll {
  685. return s.sendOldMessages(topics, since, scheduled, sub)
  686. }
  687. subscriberIDs := make([]int, 0)
  688. for _, t := range topics {
  689. subscriberIDs = append(subscriberIDs, t.Subscribe(sub))
  690. }
  691. defer func() {
  692. for i, subscriberID := range subscriberIDs {
  693. topics[i].Unsubscribe(subscriberID) // Order!
  694. }
  695. }()
  696. if err := sub(newOpenMessage(topicsStr)); err != nil { // Send out open message
  697. return err
  698. }
  699. if err := s.sendOldMessages(topics, since, scheduled, sub); err != nil {
  700. return err
  701. }
  702. for {
  703. select {
  704. case <-r.Context().Done():
  705. return nil
  706. case <-time.After(s.config.KeepaliveInterval):
  707. v.Keepalive()
  708. if err := sub(newKeepaliveMessage(topicsStr)); err != nil { // Send keepalive message
  709. return err
  710. }
  711. }
  712. }
  713. }
  714. func (s *Server) handleSubscribeWS(w http.ResponseWriter, r *http.Request, v *visitor) error {
  715. if r.Header.Get("Upgrade") != "websocket" {
  716. return errHTTPBadRequestWebSocketsUpgradeHeaderMissing
  717. }
  718. if err := v.SubscriptionAllowed(); err != nil {
  719. return errHTTPTooManyRequestsLimitSubscriptions
  720. }
  721. defer v.RemoveSubscription()
  722. topics, topicsStr, err := s.topicsFromPath(r.URL.Path)
  723. if err != nil {
  724. return err
  725. }
  726. poll, since, scheduled, filters, err := parseSubscribeParams(r)
  727. if err != nil {
  728. return err
  729. }
  730. upgrader := &websocket.Upgrader{
  731. ReadBufferSize: wsBufferSize,
  732. WriteBufferSize: wsBufferSize,
  733. CheckOrigin: func(r *http.Request) bool {
  734. return true // We're open for business!
  735. },
  736. }
  737. conn, err := upgrader.Upgrade(w, r, nil)
  738. if err != nil {
  739. return err
  740. }
  741. defer conn.Close()
  742. var wlock sync.Mutex
  743. g, ctx := errgroup.WithContext(context.Background())
  744. g.Go(func() error {
  745. pongWait := s.config.KeepaliveInterval + wsPongWait
  746. conn.SetReadLimit(wsReadLimit)
  747. if err := conn.SetReadDeadline(time.Now().Add(pongWait)); err != nil {
  748. return err
  749. }
  750. conn.SetPongHandler(func(appData string) error {
  751. return conn.SetReadDeadline(time.Now().Add(pongWait))
  752. })
  753. for {
  754. _, _, err := conn.NextReader()
  755. if err != nil {
  756. return err
  757. }
  758. }
  759. })
  760. g.Go(func() error {
  761. ping := func() error {
  762. wlock.Lock()
  763. defer wlock.Unlock()
  764. if err := conn.SetWriteDeadline(time.Now().Add(wsWriteWait)); err != nil {
  765. return err
  766. }
  767. return conn.WriteMessage(websocket.PingMessage, nil)
  768. }
  769. for {
  770. select {
  771. case <-ctx.Done():
  772. return nil
  773. case <-time.After(s.config.KeepaliveInterval):
  774. v.Keepalive()
  775. if err := ping(); err != nil {
  776. return err
  777. }
  778. }
  779. }
  780. })
  781. sub := func(msg *message) error {
  782. if !filters.Pass(msg) {
  783. return nil
  784. }
  785. wlock.Lock()
  786. defer wlock.Unlock()
  787. if err := conn.SetWriteDeadline(time.Now().Add(wsWriteWait)); err != nil {
  788. return err
  789. }
  790. return conn.WriteJSON(msg)
  791. }
  792. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  793. if poll {
  794. return s.sendOldMessages(topics, since, scheduled, sub)
  795. }
  796. subscriberIDs := make([]int, 0)
  797. for _, t := range topics {
  798. subscriberIDs = append(subscriberIDs, t.Subscribe(sub))
  799. }
  800. defer func() {
  801. for i, subscriberID := range subscriberIDs {
  802. topics[i].Unsubscribe(subscriberID) // Order!
  803. }
  804. }()
  805. if err := sub(newOpenMessage(topicsStr)); err != nil { // Send out open message
  806. return err
  807. }
  808. if err := s.sendOldMessages(topics, since, scheduled, sub); err != nil {
  809. return err
  810. }
  811. err = g.Wait()
  812. if err != nil && websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
  813. return nil // Normal closures are not errors
  814. }
  815. return err
  816. }
  817. func parseSubscribeParams(r *http.Request) (poll bool, since sinceMarker, scheduled bool, filters *queryFilter, err error) {
  818. poll = readBoolParam(r, false, "x-poll", "poll", "po")
  819. scheduled = readBoolParam(r, false, "x-scheduled", "scheduled", "sched")
  820. since, err = parseSince(r, poll)
  821. if err != nil {
  822. return
  823. }
  824. filters, err = parseQueryFilters(r)
  825. if err != nil {
  826. return
  827. }
  828. return
  829. }
  830. func (s *Server) sendOldMessages(topics []*topic, since sinceMarker, scheduled bool, sub subscriber) error {
  831. if since.IsNone() {
  832. return nil
  833. }
  834. for _, t := range topics {
  835. messages, err := s.messageCache.Messages(t.ID, since, scheduled)
  836. if err != nil {
  837. return err
  838. }
  839. for _, m := range messages {
  840. if err := sub(m); err != nil {
  841. return err
  842. }
  843. }
  844. }
  845. return nil
  846. }
  847. // parseSince returns a timestamp identifying the time span from which cached messages should be received.
  848. //
  849. // Values in the "since=..." parameter can be either a unix timestamp or a duration (e.g. 12h), or
  850. // "all" for all messages.
  851. func parseSince(r *http.Request, poll bool) (sinceMarker, error) {
  852. since := readParam(r, "x-since", "since", "si")
  853. // Easy cases (empty, all, none)
  854. if since == "" {
  855. if poll {
  856. return sinceAllMessages, nil
  857. }
  858. return sinceNoMessages, nil
  859. } else if since == "all" {
  860. return sinceAllMessages, nil
  861. } else if since == "none" {
  862. return sinceNoMessages, nil
  863. }
  864. // ID/timestamp
  865. parts := strings.Split(since, "/")
  866. if len(parts) == 2 && validMessageID(parts[0]) && validUnixTimestamp(parts[1]) {
  867. t, _ := toUnixTimestamp(parts[1])
  868. return newSince(parts[0], t), nil
  869. }
  870. // ID, timestamp, duration
  871. if validMessageID(since) {
  872. return newSinceID(since), nil
  873. } else if s, err := strconv.ParseInt(since, 10, 64); err == nil {
  874. return newSinceTime(s), nil
  875. } else if d, err := time.ParseDuration(since); err == nil {
  876. return newSinceTime(time.Now().Add(-1 * d).Unix()), nil
  877. }
  878. return sinceNoMessages, errHTTPBadRequestSinceInvalid
  879. }
  880. func (s *Server) handleOptions(w http.ResponseWriter, _ *http.Request) error {
  881. w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, POST")
  882. w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
  883. w.Header().Set("Access-Control-Allow-Headers", "*") // CORS, allow auth via JS // FIXME is this terrible?
  884. return nil
  885. }
  886. func (s *Server) topicAndMessageIDFromPath(path string) (*topic, string, error) {
  887. parts := strings.Split(path, "/")
  888. if len(parts) != 2 && len(parts) != 3 {
  889. return nil, "", errHTTPBadRequestTopicInvalid
  890. }
  891. topics, err := s.topicsFromIDs(parts[1])
  892. if err != nil {
  893. return nil, "", err
  894. }
  895. messageID := ""
  896. if len(parts) == 3 && len(parts[2]) == messageIDLength {
  897. messageID = parts[2]
  898. }
  899. return topics[0], messageID, nil
  900. }
  901. func (s *Server) topicsFromPath(path string) ([]*topic, string, error) {
  902. parts := strings.Split(path, "/")
  903. if len(parts) < 2 {
  904. return nil, "", errHTTPBadRequestTopicInvalid
  905. }
  906. topicIDs := util.SplitNoEmpty(parts[1], ",")
  907. topics, err := s.topicsFromIDs(topicIDs...)
  908. if err != nil {
  909. return nil, "", errHTTPBadRequestTopicInvalid
  910. }
  911. return topics, parts[1], nil
  912. }
  913. func (s *Server) topicsFromIDs(ids ...string) ([]*topic, error) {
  914. s.mu.Lock()
  915. defer s.mu.Unlock()
  916. topics := make([]*topic, 0)
  917. for _, id := range ids {
  918. if util.InStringList(disallowedTopics, id) {
  919. return nil, errHTTPBadRequestTopicDisallowed
  920. }
  921. if _, ok := s.topics[id]; !ok {
  922. if len(s.topics) >= s.config.TotalTopicLimit {
  923. return nil, errHTTPTooManyRequestsLimitTotalTopics
  924. }
  925. s.topics[id] = newTopic(id)
  926. }
  927. topics = append(topics, s.topics[id])
  928. }
  929. return topics, nil
  930. }
  931. func (s *Server) updateStatsAndPrune() {
  932. s.mu.Lock()
  933. defer s.mu.Unlock()
  934. // Expire visitors from rate visitors map
  935. for ip, v := range s.visitors {
  936. if v.Stale() {
  937. delete(s.visitors, ip)
  938. }
  939. }
  940. // Delete expired attachments
  941. if s.fileCache != nil {
  942. ids, err := s.messageCache.AttachmentsExpired()
  943. if err == nil {
  944. if err := s.fileCache.Remove(ids...); err != nil {
  945. log.Printf("error while deleting attachments: %s", err.Error())
  946. }
  947. } else {
  948. log.Printf("error retrieving expired attachments: %s", err.Error())
  949. }
  950. }
  951. // Prune message cache
  952. olderThan := time.Now().Add(-1 * s.config.CacheDuration)
  953. if err := s.messageCache.Prune(olderThan); err != nil {
  954. log.Printf("error pruning cache: %s", err.Error())
  955. }
  956. // Prune old topics, remove subscriptions without subscribers
  957. var subscribers, messages int
  958. for _, t := range s.topics {
  959. subs := t.Subscribers()
  960. msgs, err := s.messageCache.MessageCount(t.ID)
  961. if err != nil {
  962. log.Printf("cannot get stats for topic %s: %s", t.ID, err.Error())
  963. continue
  964. }
  965. if msgs == 0 && subs == 0 {
  966. delete(s.topics, t.ID)
  967. continue
  968. }
  969. subscribers += subs
  970. messages += msgs
  971. }
  972. // Mail stats
  973. var mailSuccess, mailFailure int64
  974. if s.smtpBackend != nil {
  975. mailSuccess, mailFailure = s.smtpBackend.Counts()
  976. }
  977. // Print stats
  978. log.Printf("Stats: %d message(s) published, %d in cache, %d successful mails, %d failed, %d topic(s) active, %d subscriber(s), %d visitor(s)",
  979. s.messages, messages, mailSuccess, mailFailure, len(s.topics), subscribers, len(s.visitors))
  980. }
  981. func (s *Server) runSMTPServer() error {
  982. sub := func(m *message) error {
  983. url := fmt.Sprintf("%s/%s", s.config.BaseURL, m.Topic)
  984. req, err := http.NewRequest("PUT", url, strings.NewReader(m.Message))
  985. if err != nil {
  986. return err
  987. }
  988. if m.Title != "" {
  989. req.Header.Set("Title", m.Title)
  990. }
  991. rr := httptest.NewRecorder()
  992. s.handle(rr, req)
  993. if rr.Code != http.StatusOK {
  994. return errors.New("error: " + rr.Body.String())
  995. }
  996. return nil
  997. }
  998. s.smtpBackend = newMailBackend(s.config, sub)
  999. s.smtpServer = smtp.NewServer(s.smtpBackend)
  1000. s.smtpServer.Addr = s.config.SMTPServerListen
  1001. s.smtpServer.Domain = s.config.SMTPServerDomain
  1002. s.smtpServer.ReadTimeout = 10 * time.Second
  1003. s.smtpServer.WriteTimeout = 10 * time.Second
  1004. s.smtpServer.MaxMessageBytes = 1024 * 1024 // Must be much larger than message size (headers, multipart, etc.)
  1005. s.smtpServer.MaxRecipients = 1
  1006. s.smtpServer.AllowInsecureAuth = true
  1007. return s.smtpServer.ListenAndServe()
  1008. }
  1009. func (s *Server) runManager() {
  1010. for {
  1011. select {
  1012. case <-time.After(s.config.ManagerInterval):
  1013. s.updateStatsAndPrune()
  1014. case <-s.closeChan:
  1015. return
  1016. }
  1017. }
  1018. }
  1019. func (s *Server) runAtSender() {
  1020. for {
  1021. select {
  1022. case <-time.After(s.config.AtSenderInterval):
  1023. if err := s.sendDelayedMessages(); err != nil {
  1024. log.Printf("error sending scheduled messages: %s", err.Error())
  1025. }
  1026. case <-s.closeChan:
  1027. return
  1028. }
  1029. }
  1030. }
  1031. func (s *Server) runFirebaseKeepaliver() {
  1032. if s.firebase == nil {
  1033. return
  1034. }
  1035. for {
  1036. select {
  1037. case <-time.After(s.config.FirebaseKeepaliveInterval):
  1038. if err := s.firebase(newKeepaliveMessage(firebaseControlTopic)); err != nil {
  1039. log.Printf("error sending Firebase keepalive message: %s", err.Error())
  1040. }
  1041. case <-s.closeChan:
  1042. return
  1043. }
  1044. }
  1045. }
  1046. func (s *Server) sendDelayedMessages() error {
  1047. s.mu.Lock()
  1048. defer s.mu.Unlock()
  1049. messages, err := s.messageCache.MessagesDue()
  1050. if err != nil {
  1051. return err
  1052. }
  1053. for _, m := range messages {
  1054. t, ok := s.topics[m.Topic] // If no subscribers, just mark message as published
  1055. if ok {
  1056. if err := t.Publish(m); err != nil {
  1057. log.Printf("unable to publish message %s to topic %s: %v", m.ID, m.Topic, err.Error())
  1058. }
  1059. }
  1060. if s.firebase != nil { // Firebase subscribers may not show up in topics map
  1061. if err := s.firebase(m); err != nil {
  1062. log.Printf("unable to publish to Firebase: %v", err.Error())
  1063. }
  1064. }
  1065. if err := s.messageCache.MarkPublished(m); err != nil {
  1066. return err
  1067. }
  1068. }
  1069. return nil
  1070. }
  1071. func (s *Server) limitRequests(next handleFunc) handleFunc {
  1072. return func(w http.ResponseWriter, r *http.Request, v *visitor) error {
  1073. if util.InStringList(s.config.VisitorRequestExemptIPAddrs, v.ip) {
  1074. return next(w, r, v)
  1075. } else if err := v.RequestAllowed(); err != nil {
  1076. return errHTTPTooManyRequestsLimitRequests
  1077. }
  1078. return next(w, r, v)
  1079. }
  1080. }
  1081. // transformBodyJSON peaks the request body, reads the JSON, and converts it to headers
  1082. // before passing it on to the next handler. This is meant to be used in combination with handlePublish.
  1083. func (s *Server) transformBodyJSON(next handleFunc) handleFunc {
  1084. return func(w http.ResponseWriter, r *http.Request, v *visitor) error {
  1085. body, err := util.Peak(r.Body, s.config.MessageLimit)
  1086. if err != nil {
  1087. return err
  1088. }
  1089. defer r.Body.Close()
  1090. var m publishMessage
  1091. if err := json.NewDecoder(body).Decode(&m); err != nil {
  1092. return errHTTPBadRequestJSONInvalid
  1093. }
  1094. if !topicRegex.MatchString(m.Topic) {
  1095. return errHTTPBadRequestTopicInvalid
  1096. }
  1097. if m.Message == "" {
  1098. m.Message = emptyMessageBody
  1099. }
  1100. r.URL.Path = "/" + m.Topic
  1101. r.Body = io.NopCloser(strings.NewReader(m.Message))
  1102. if m.Title != "" {
  1103. r.Header.Set("X-Title", m.Title)
  1104. }
  1105. if m.Priority != 0 {
  1106. r.Header.Set("X-Priority", fmt.Sprintf("%d", m.Priority))
  1107. }
  1108. if m.Tags != nil && len(m.Tags) > 0 {
  1109. r.Header.Set("X-Tags", strings.Join(m.Tags, ","))
  1110. }
  1111. if m.Attach != "" {
  1112. r.Header.Set("X-Attach", m.Attach)
  1113. }
  1114. if m.Filename != "" {
  1115. r.Header.Set("X-Filename", m.Filename)
  1116. }
  1117. if m.Click != "" {
  1118. r.Header.Set("X-Click", m.Click)
  1119. }
  1120. return next(w, r, v)
  1121. }
  1122. }
  1123. func (s *Server) authWrite(next handleFunc) handleFunc {
  1124. return s.withAuth(next, auth.PermissionWrite)
  1125. }
  1126. func (s *Server) authRead(next handleFunc) handleFunc {
  1127. return s.withAuth(next, auth.PermissionRead)
  1128. }
  1129. func (s *Server) withAuth(next handleFunc, perm auth.Permission) handleFunc {
  1130. return func(w http.ResponseWriter, r *http.Request, v *visitor) error {
  1131. if s.auth == nil {
  1132. return next(w, r, v)
  1133. }
  1134. topics, _, err := s.topicsFromPath(r.URL.Path)
  1135. if err != nil {
  1136. return err
  1137. }
  1138. var user *auth.User // may stay nil if no auth header!
  1139. username, password, ok := extractUserPass(r)
  1140. if ok {
  1141. if user, err = s.auth.Authenticate(username, password); err != nil {
  1142. log.Printf("authentication failed: %s", err.Error())
  1143. return errHTTPUnauthorized
  1144. }
  1145. }
  1146. for _, t := range topics {
  1147. if err := s.auth.Authorize(user, t.ID, perm); err != nil {
  1148. log.Printf("unauthorized: %s", err.Error())
  1149. return errHTTPForbidden
  1150. }
  1151. }
  1152. return next(w, r, v)
  1153. }
  1154. }
  1155. // extractUserPass reads the username/password from the basic auth header (Authorization: Basic ...),
  1156. // or from the ?auth=... query param. The latter is required only to support the WebSocket JavaScript
  1157. // class, which does not support passing headers during the initial request. The auth query param
  1158. // is effectively double base64 encoded. Its format is base64(Basic base64(user:pass)).
  1159. func extractUserPass(r *http.Request) (username string, password string, ok bool) {
  1160. username, password, ok = r.BasicAuth()
  1161. if ok {
  1162. return
  1163. }
  1164. authParam := readQueryParam(r, "authorization", "auth")
  1165. if authParam != "" {
  1166. a, err := base64.RawURLEncoding.DecodeString(authParam)
  1167. if err != nil {
  1168. return
  1169. }
  1170. r.Header.Set("Authorization", string(a))
  1171. return r.BasicAuth()
  1172. }
  1173. return
  1174. }
  1175. // visitor creates or retrieves a rate.Limiter for the given visitor.
  1176. // This function was taken from https://www.alexedwards.net/blog/how-to-rate-limit-http-requests (MIT).
  1177. func (s *Server) visitor(r *http.Request) *visitor {
  1178. s.mu.Lock()
  1179. defer s.mu.Unlock()
  1180. remoteAddr := r.RemoteAddr
  1181. ip, _, err := net.SplitHostPort(remoteAddr)
  1182. if err != nil {
  1183. ip = remoteAddr // This should not happen in real life; only in tests.
  1184. }
  1185. if s.config.BehindProxy && r.Header.Get("X-Forwarded-For") != "" {
  1186. ip = r.Header.Get("X-Forwarded-For")
  1187. }
  1188. v, exists := s.visitors[ip]
  1189. if !exists {
  1190. s.visitors[ip] = newVisitor(s.config, ip)
  1191. return s.visitors[ip]
  1192. }
  1193. v.Keepalive()
  1194. return v
  1195. }