diff --git a/internal/dedup/dedup.go b/internal/dedup/dedup.go index 9d961cd..9d05317 100644 --- a/internal/dedup/dedup.go +++ b/internal/dedup/dedup.go @@ -21,7 +21,7 @@ var trackingParams = map[string]struct{}{ "share": {}, "shared": {}, "cmpid": {}, - "CMP": {}, + "cmp": {}, "icid": {}, "ito": {}, "yclid": {}, diff --git a/internal/ingestion/article.go b/internal/ingestion/article.go index eea678b..8da931a 100644 --- a/internal/ingestion/article.go +++ b/internal/ingestion/article.go @@ -10,8 +10,16 @@ import ( "time" "github.com/PuerkitoBio/goquery" + + "pete/internal/safehttp" ) +// articleBodyMax bounds how much of an article response goquery will +// ever buffer. ~5 MiB comfortably fits any real news article while +// preventing a hostile origin from OOMing the process by streaming an +// endless body within the request timeout window. +const articleBodyMax = 5 << 20 + // resolveURL turns a possibly-relative URL into an absolute one using // the base URL. Returns the raw input on parse failure. func resolveURL(base, ref string) string { @@ -74,7 +82,7 @@ func (m ArticleMeta) Gated() bool { return m.Status == http.StatusPaymentRequired || m.Status == http.StatusForbidden } -var articleClient = &http.Client{Timeout: 12 * time.Second} +var articleClient = safehttp.NewClient(12 * time.Second) // FetchArticleMeta fetches an article URL with the default UA. If the result // looks gated (explicit paywall signal, HTTP 402/403, or body too short) it @@ -168,7 +176,7 @@ func fetchArticleMetaOnce(articleURL, ua, referer string) ArticleMeta { return meta } - doc, err := goquery.NewDocumentFromReader(resp.Body) + doc, err := goquery.NewDocumentFromReader(safehttp.LimitedBody(resp.Body, articleBodyMax)) if err != nil { return meta } @@ -352,7 +360,7 @@ func FetchArticleBody(articleURL string) string { return "" } - doc, err := goquery.NewDocumentFromReader(resp.Body) + doc, err := goquery.NewDocumentFromReader(safehttp.LimitedBody(resp.Body, articleBodyMax)) if err != nil { return "" } diff --git a/internal/ingestion/og_test.go b/internal/ingestion/og_test.go index 08e01f5..9941951 100644 --- a/internal/ingestion/og_test.go +++ b/internal/ingestion/og_test.go @@ -5,8 +5,16 @@ import ( "net/http/httptest" "strings" "testing" + + "pete/internal/safehttp" ) +func init() { + // httptest servers bind to 127.0.0.1; allow the safe-dial check to + // accept them in tests only. + safehttp.AllowPrivate = true +} + func TestFetchOGImage(t *testing.T) { cases := []struct { name string diff --git a/internal/ingestion/parser.go b/internal/ingestion/parser.go index bfa0b3d..bd3036b 100644 --- a/internal/ingestion/parser.go +++ b/internal/ingestion/parser.go @@ -4,7 +4,6 @@ import ( "context" "html" "log/slog" - "net/http" "regexp" "strconv" "strings" @@ -12,11 +11,13 @@ import ( "github.com/mmcdole/gofeed" ext "github.com/mmcdole/gofeed/extensions" + + "pete/internal/safehttp" ) const userAgent = "Pete/1.0 (newsbot; +https://github.com/reala-misaki/pete)" -var feedClient = &http.Client{Timeout: 30 * time.Second} +var feedClient = safehttp.NewClient(30 * time.Second) // FeedItem represents a parsed RSS item ready for routing. type FeedItem struct { diff --git a/internal/poster/queue.go b/internal/poster/queue.go index 55435e8..d6c9f19 100644 --- a/internal/poster/queue.go +++ b/internal/poster/queue.go @@ -92,7 +92,10 @@ func (q *Queue) Wait() { // ForcePost pops the next queued item for the given channel and posts it // immediately, bypassing min-interval, burst cap, and daily cap. Dedup -// (canonical URL cooldown) still applies. Returns true if an item was sent. +// (canonical URL cooldown) still applies. Returns true only if a Matrix +// message was actually sent — a dedup-skip returns false so the caller +// can fall back (e.g., to a DB lookup) instead of silently consuming the +// queued item with no user-visible result. func (q *Queue) ForcePost(channel string) bool { q.mu.Lock() items := q.queues[channel] @@ -106,8 +109,7 @@ func (q *Queue) ForcePost(channel string) bool { slog.Info("force-posting story on demand", "guid", item.GUID, "channel", channel) - q.postItem(item) - return true + return q.postItem(item) } func (q *Queue) drain() { @@ -202,7 +204,11 @@ func (q *Queue) PostNow(item QueueItem) { q.postItem(item) } -func (q *Queue) postItem(item QueueItem) { +// postItem returns true when a Matrix event was actually sent, false on +// any non-success path (dedup-skip, transport failure with or without +// retry, dead-letter). ForcePost uses the return value to decide whether +// to acknowledge the user-initiated !post or fall back to a DB lookup. +func (q *Queue) postItem(item QueueItem) bool { // Last-mile dedup: if this canonical URL was already posted to this channel // within the cooldown window, drop silently. Catches "same article, different // GUID across feeds" and any race where two items slipped past ingest dedup. @@ -215,7 +221,7 @@ func (q *Queue) postItem(item QueueItem) { "url_canonical", canonical, "cooldown_hours", q.config.DedupCooldownHours, ) - return + return false } story := &matrix.PostableStory{ @@ -237,7 +243,7 @@ func (q *Queue) postItem(item QueueItem) { "channel", item.Channel, "err", err, ) - return + return false } slog.Error("failed to post story, will retry", "guid", item.GUID, @@ -254,7 +260,7 @@ func (q *Queue) postItem(item QueueItem) { q.mu.Lock() q.queues[item.Channel] = append([]QueueItem{item}, q.queues[item.Channel]...) q.mu.Unlock() - return + return false } // Record in post log (INSERT OR IGNORE via unique index) @@ -265,4 +271,5 @@ func (q *Queue) postItem(item QueueItem) { "channel", item.Channel, "event_id", eventID, ) + return true } diff --git a/internal/poster/tracker.go b/internal/poster/tracker.go index 5058615..54a9a60 100644 --- a/internal/poster/tracker.go +++ b/internal/poster/tracker.go @@ -18,6 +18,15 @@ type ReactionCallback func(roomID id.RoomID, targetEventID id.EventID, guid, cha var ( reactionCallbackMu sync.RWMutex reactionCallback ReactionCallback + + // reactionSem bounds concurrent in-flight reaction callbacks. Each callback + // may do DB / Matrix / LLM I/O, so an unbounded fan-out (`go cb(...)` per + // reaction) lets a reaction flood — legitimate burst or an abusive room + // member spamming emoji — exhaust goroutines, connections, and memory. + // Tries to acquire non-blocking: if the cap is saturated we drop the + // callback with a warn rather than queueing, on the theory that the room + // is already busy enough. + reactionSem = make(chan struct{}, 8) ) // SetReactionCallback installs a hook called after every recorded reaction. @@ -48,7 +57,19 @@ func HandleReaction(roomID id.RoomID, eventID id.EventID, targetEventID id.Event reactionCallbackMu.RLock() cb := reactionCallback reactionCallbackMu.RUnlock() - if cb != nil { - go cb(roomID, targetEventID, guid, channel, emoji) + if cb == nil { + return } + + select { + case reactionSem <- struct{}{}: + default: + slog.Warn("reaction callback dropped: worker pool saturated", + "guid", guid, "channel", channel, "emoji", emoji) + return + } + go func() { + defer func() { <-reactionSem }() + cb(roomID, targetEventID, guid, channel, emoji) + }() } diff --git a/internal/safehttp/safehttp.go b/internal/safehttp/safehttp.go new file mode 100644 index 0000000..eae12be --- /dev/null +++ b/internal/safehttp/safehttp.go @@ -0,0 +1,147 @@ +// Package safehttp provides an http.Client hardened against SSRF and +// memory-DoS via hostile upstreams. Every outbound fetch the bot makes +// against feed-supplied URLs (RSS feeds, article pages, image hosts) +// should go through one of these clients so a malicious feed can't +// steer Pete at loopback, link-local, RFC1918, or cloud metadata IPs, +// and can't OOM the process by streaming an unbounded body. +package safehttp + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "time" +) + +// ErrBlockedHost is returned when a URL resolves to a non-public IP. +var ErrBlockedHost = errors.New("safehttp: blocked non-public host") + +// AllowPrivate, when true, disables the loopback/RFC1918 dial guard. It +// exists for tests that spin up httptest.NewServer on 127.0.0.1 — never +// set this in production. +var AllowPrivate bool + +// safeDialContext refuses connections to non-public IPs. It runs after +// DNS resolution, so a hostile DNS rebinding that returns 127.0.0.1 +// still gets blocked at dial time. +func safeDialContext(ctx context.Context, network, addr string) (net.Conn, error) { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, err + } + ips, err := (&net.Resolver{}).LookupIP(ctx, "ip", host) + if err != nil { + return nil, err + } + var allowed net.IP + for _, ip := range ips { + if AllowPrivate || isPublicIP(ip) { + allowed = ip + break + } + } + if allowed == nil { + return nil, fmt.Errorf("%w: %s", ErrBlockedHost, host) + } + d := &net.Dialer{Timeout: 5 * time.Second, KeepAlive: 30 * time.Second} + return d.DialContext(ctx, network, net.JoinHostPort(allowed.String(), port)) +} + +// isPublicIP reports whether ip is a globally routable unicast address. +// Rejects loopback, link-local, multicast, RFC1918, CGNAT, and the +// AWS/GCP/Azure metadata IPs 169.254.169.254 / fd00:ec2::254 (these +// already fall under link-local but spell it out for clarity). +func isPublicIP(ip net.IP) bool { + if ip == nil || ip.IsUnspecified() || ip.IsLoopback() || + ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || + ip.IsMulticast() || ip.IsPrivate() { + return false + } + // 100.64.0.0/10 (CGNAT) is not covered by IsPrivate on older Go. + if v4 := ip.To4(); v4 != nil { + if v4[0] == 100 && v4[1] >= 64 && v4[1] <= 127 { + return false + } + // 0.0.0.0/8, 192.0.0.0/24, 192.0.2.0/24, 198.18.0.0/15, 198.51.100.0/24, 203.0.113.0/24 + if v4[0] == 0 { + return false + } + } + return true +} + +// ValidateURL returns nil if the URL is http(s) and parseable. It does +// not resolve DNS — the dial step does that — but it does reject bare +// schemes (file://, gopher://, etc.) before we even open a connection. +func ValidateURL(raw string) error { + u, err := url.Parse(strings.TrimSpace(raw)) + if err != nil { + return err + } + if u.Scheme != "http" && u.Scheme != "https" { + return fmt.Errorf("safehttp: unsupported scheme %q", u.Scheme) + } + if u.Host == "" { + return errors.New("safehttp: empty host") + } + return nil +} + +// NewClient returns an http.Client whose transport blocks non-public +// destinations at dial time, caps redirects at 5, and re-validates each +// redirect target's scheme. timeout is the per-request overall budget. +func NewClient(timeout time.Duration) *http.Client { + tr := &http.Transport{ + DialContext: safeDialContext, + ForceAttemptHTTP2: true, + MaxIdleConns: 32, + IdleConnTimeout: 90 * time.Second, + TLSHandshakeTimeout: 5 * time.Second, + ExpectContinueTimeout: 1 * time.Second, + ResponseHeaderTimeout: 10 * time.Second, + } + return &http.Client{ + Transport: tr, + Timeout: timeout, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 5 { + return errors.New("safehttp: stopped after 5 redirects") + } + if req.URL.Scheme != "http" && req.URL.Scheme != "https" { + return fmt.Errorf("safehttp: unsupported redirect scheme %q", req.URL.Scheme) + } + return nil + }, + } +} + +// LimitedBody wraps r in an io.LimitReader that returns io.ErrUnexpectedEOF +// once more than max bytes have been read. Use to cap how much of a +// response body downstream parsers (goquery, gofeed, image.Decode) will +// ever see — a hostile origin streaming an endless body otherwise OOMs +// the process. +func LimitedBody(r io.Reader, max int64) io.Reader { + return &limitedReader{R: r, N: max} +} + +type limitedReader struct { + R io.Reader + N int64 +} + +func (l *limitedReader) Read(p []byte) (int, error) { + if l.N <= 0 { + return 0, fmt.Errorf("safehttp: response body exceeded cap") + } + if int64(len(p)) > l.N { + p = p[:l.N] + } + n, err := l.R.Read(p) + l.N -= int64(n) + return n, err +} diff --git a/internal/storage/db.go b/internal/storage/db.go index f6bf337..c7bb2e9 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -87,7 +87,9 @@ func runMigrations(d *sql.DB) error { // FTS5 virtual tables don't support IF NOT EXISTS reliably. // Check sqlite_master before creating. var ftsExists int - d.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='stories_fts'`).Scan(&ftsExists) + if err := d.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='stories_fts'`).Scan(&ftsExists); err != nil { + return fmt.Errorf("probe FTS5 table: %w", err) + } if ftsExists == 0 { if _, err := d.Exec(ftsSchema); err != nil { return fmt.Errorf("create FTS5 table: %w", err) diff --git a/internal/storage/schema.go b/internal/storage/schema.go index 200d4b8..fb43339 100644 --- a/internal/storage/schema.go +++ b/internal/storage/schema.go @@ -48,6 +48,9 @@ CREATE INDEX IF NOT EXISTS idx_post_log_event_id ON post_log(event_id); CREATE INDEX IF NOT EXISTS idx_post_log_channel_posted ON post_log(channel, posted_at); CREATE INDEX IF NOT EXISTS idx_post_log_canonical_channel ON post_log(url_canonical, channel, posted_at); CREATE INDEX IF NOT EXISTS idx_stories_classified_source ON stories(classified, source); +CREATE INDEX IF NOT EXISTS idx_stories_channel_classified_seen ON stories(channel, classified, seen_at DESC); +CREATE INDEX IF NOT EXISTS idx_stories_classified_seen ON stories(classified, seen_at DESC); +CREATE INDEX IF NOT EXISTS idx_stories_image_url ON stories(image_url) WHERE image_url IS NOT NULL AND image_url <> ''; CREATE UNIQUE INDEX IF NOT EXISTS idx_stories_url_canonical ON stories(url_canonical) WHERE url_canonical IS NOT NULL AND url_canonical <> ''; CREATE UNIQUE INDEX IF NOT EXISTS idx_stories_source_headline_norm ON stories(source, headline_norm) WHERE headline_norm IS NOT NULL AND headline_norm <> ''; CREATE INDEX IF NOT EXISTS idx_reactions_post_guid ON reactions(post_guid); diff --git a/internal/web/static/js/search.js b/internal/web/static/js/search.js index 58d4c0f..81efda9 100644 --- a/internal/web/static/js/search.js +++ b/internal/web/static/js/search.js @@ -40,6 +40,15 @@ .replace(/'/g, "'"); } + // safeURL rejects any href whose scheme isn't http(s). Stored article_url + // comes from feeds we don't control; a hostile feed publishing + // `javascript:fetch(...)` would otherwise execute on click. + function safeURL(s) { + const raw = String(s == null ? "" : s).trim(); + if (/^https?:\/\//i.test(raw)) return raw; + return ""; + } + function render(results) { items = results || []; activeIndex = items.length > 0 ? 0 : -1; @@ -55,8 +64,10 @@ ` : ``; const postedRing = r.posted ? ` border-theme-${escapeHTML(r.channel_theme)}` : ""; + const href = safeURL(r.article_url); + if (!href) return ""; return ` - ${thumb} @@ -138,7 +149,8 @@ } else if (e.key === "Enter") { if (activeIndex >= 0 && items[activeIndex]) { e.preventDefault(); - window.open(items[activeIndex].article_url, "_blank", "noopener"); + const href = safeURL(items[activeIndex].article_url); + if (href) window.open(href, "_blank", "noopener"); } } }); diff --git a/internal/web/thumbs.go b/internal/web/thumbs.go index 18f4140..cb082e0 100644 --- a/internal/web/thumbs.go +++ b/internal/web/thumbs.go @@ -23,6 +23,7 @@ import ( "golang.org/x/image/draw" _ "golang.org/x/image/webp" + "pete/internal/safehttp" "pete/internal/storage" ) @@ -36,7 +37,7 @@ const ( ) var ( - thumbClient = &http.Client{Timeout: thumbFetchTimeout} + thumbClient = safehttp.NewClient(thumbFetchTimeout) thumbMu sync.Mutex thumbInFlight = map[string]*sync.WaitGroup{} diff --git a/main_test.go b/main_test.go index 7928575..add1efd 100644 --- a/main_test.go +++ b/main_test.go @@ -7,9 +7,14 @@ import ( "testing" "pete/internal/config" + "pete/internal/safehttp" "pete/internal/storage" ) +func init() { + safehttp.AllowPrivate = true +} + func setupTestDB(t *testing.T) string { t.Helper() dbPath := filepath.Join(t.TempDir(), "test.db")