Fix SSRF, XSS, dedup, force-post, and DB hot-path issues

- Add internal/safehttp: hardened HTTP client (DNS-resolved dial guard
  blocking loopback/RFC1918/CGNAT/link-local, redirect re-validation,
  body-size cap) and rewire article/feed/thumb clients through it
- Cap goquery body at 5 MiB so a hostile origin can't OOM the process
- search.js: reject non-http(s) hrefs to block stored XSS via javascript:
- dedup: tracking-param key "CMP" was unreachable (lookup lowercases);
  fixed to "cmp" so CMP= is actually stripped from canonical URLs
- ForcePost: postItem now returns bool; on dedup-skip ForcePost returns
  false so !post falls back to DB lookup instead of silently consuming
- Bound reaction callbacks behind an 8-slot semaphore; drop overflow
- Add stories indexes on (channel, classified, seen_at DESC),
  (classified, seen_at DESC), and partial image_url to kill full scans
  in IsKnownImageURL and ORDER BY seen_at hot paths
- Surface FTS5 probe Scan error instead of swallowing it

internal/dedup/dedup.go

@@ -21,7 +21,7 @@ var trackingParams = map[string]struct{}{
 	"share":       {},
 	"shared":      {},
 	"cmpid":       {},
-	"CMP":         {},
+	"cmp":         {},
 	"icid":        {},
 	"ito":         {},
 	"yclid":       {},

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 ""
 	}

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

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 {

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
 }

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)
+	}()
 }

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
+}

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)

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);

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 @@
            </div>`
         : `<div class="hidden sm:flex w-20 h-20 shrink-0 items-center justify-center rounded-2xl bg-[color:var(--ink)]/5 text-2xl">${escapeHTML(r.channel_emoji || "·")}</div>`;
       const postedRing = r.posted ? ` border-theme-${escapeHTML(r.channel_theme)}` : "";
+      const href = safeURL(r.article_url);
+      if (!href) return "";
       return `
-        <a href="${escapeHTML(r.article_url)}" target="_blank" rel="noopener noreferrer"
+        <a href="${escapeHTML(href)}" target="_blank" rel="noopener noreferrer"
            data-idx="${i}"
            class="search-result group flex items-start gap-4 rounded-2xl p-3 transition border-2 border-transparent hover:bg-[color:var(--ink)]/5${postedRing}">
           ${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");
       }
     }
   });

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{}

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")