package chat
import (
"context"
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/events"
)
// fakeFossil is a fossil server's chat surface and nothing else:
// /chat-poll answers whatever the test queued, /chat-send records what
// arrived. It enforces the two gates a real fossil enforces — HTTP
// Basic and a same-origin Referer — so a bridge that forgets either
// fails here rather than in production.
type fakeFossil struct {
mu sync.Mutex
queue []fossilMsg // handed out by /chat-poll, oldest first
sent []string // the msg= of every /chat-send
polls []string // the name= of every /chat-poll
fail int // this many polls answer 500 before any succeed
noAuth bool // answer fossil's permissions error instead
srv *httptest.Server
pollHit chan struct{}
}
func newFakeFossil(t *testing.T) *fakeFossil {
t.Helper()
f := &fakeFossil{pollHit: make(chan struct{}, 64)}
mux := http.NewServeMux()
mux.HandleFunc("/chat-poll", func(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
f.polls = append(f.polls, r.URL.Query().Get("name"))
fail, noAuth := f.fail, f.noAuth
if fail > 0 {
f.fail--
}
f.mu.Unlock()
select {
case f.pollHit <- struct{}{}:
default:
}
if !f.authed(r) {
http.Error(w, "no", http.StatusUnauthorized)
return
}
if fail > 0 {
http.Error(w, "boom", http.StatusInternalServerError)
return
}
if noAuth {
// Fossil answers a refusal INSIDE a 200 JSON envelope.
fmt.Fprint(w, `{"msgs":[{"isError":true,"xfrom":null,`+
`"xmsg":"Missing permissions or not logged in. Try <a href='/login'>logging in</a>."}]}`)
return
}
since := r.URL.Query().Get("name")
f.mu.Lock()
var out []fossilMsg
for _, m := range f.queue {
if fmt.Sprint(m.MsgID) > since || since == "0" || since == "" {
out = append(out, m)
}
}
f.queue = nil
f.mu.Unlock()
_ = json.NewEncoder(w).Encode(fossilPoll{Msgs: out})
})
mux.HandleFunc("/chat-send", func(w http.ResponseWriter, r *http.Request) {
if !f.authed(r) {
http.Error(w, "no", http.StatusUnauthorized)
return
}
_ = r.ParseForm()
f.mu.Lock()
f.sent = append(f.sent, r.FormValue("msg"))
f.mu.Unlock()
})
f.srv = httptest.NewServer(mux)
t.Cleanup(f.srv.Close)
return f
}
// authed is both gates: the credential and the Referer that makes the
// request CSRF-safe in fossil's eyes.
func (f *fakeFossil) authed(r *http.Request) bool {
user, pass, ok := r.BasicAuth()
if !ok || user != "bridgebot" || pass != "s3cret" {
return false
}
return strings.HasPrefix(r.Header.Get("Referer"), f.srv.URL)
}
func (f *fakeFossil) push(msgs ...fossilMsg) {
f.mu.Lock()
f.queue = append(f.queue, msgs...)
f.mu.Unlock()
}
func (f *fakeFossil) sentMsgs() []string {
f.mu.Lock()
defer f.mu.Unlock()
return append([]string(nil), f.sent...)
}
// memBridgeStore is the cursor and msgid map in memory.
type memBridgeStore struct {
mu sync.Mutex
cursor int64
lastEr string
seqOf map[int64]uint64
}
func newMemStore() *memBridgeStore { return &memBridgeStore{seqOf: map[int64]uint64{}} }
func (m *memBridgeStore) SetChatBridgeCursor(_ string, msgid int64, _ time.Time) error {
m.mu.Lock()
defer m.mu.Unlock()
m.cursor, m.lastEr = msgid, ""
return nil
}
func (m *memBridgeStore) RecordChatBridgeError(_ string, err error) error {
m.mu.Lock()
defer m.mu.Unlock()
if err != nil {
m.lastEr = err.Error()
}
return nil
}
func (m *memBridgeStore) MapChatBridgeMsg(_ string, msgid int64, seq uint64) error {
m.mu.Lock()
defer m.mu.Unlock()
m.seqOf[msgid] = seq
return nil
}
func (m *memBridgeStore) ChatBridgeSeq(_ string, msgid int64) (uint64, bool) {
m.mu.Lock()
defer m.mu.Unlock()
seq, ok := m.seqOf[msgid]
return seq, ok
}
func (m *memBridgeStore) err() string {
m.mu.Lock()
defer m.mu.Unlock()
return m.lastEr
}
// bridgeFixture is a real room over embedded NATS plus a fake fossil.
func bridgeFixture(t *testing.T) (*Room, *fakeFossil, *memBridgeStore, *Bridge) {
t.Helper()
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
r := room(t, ev)
f := newFakeFossil(t)
store := newMemStore()
b := NewBridge(r, store, BridgeConfig{
Repo: "alpha", Code: "codeA", URL: f.srv.URL, User: "bridgebot", Secret: "s3cret",
// httptest listens on loopback, which the guard refuses by
// default — the same opt-in an operator makes for a fossil on
// the same host.
AllowPrivate: true,
})
// The poll must not hang out its production timeout in a test.
b.Client = bridgeClient(5*time.Second, true)
b.SendClient = bridgeClient(5*time.Second, true)
return r, f, store, b
}
// waitFor polls cond until it holds or the test's patience runs out.
func waitFor(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
if cond() {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("timed out waiting for %s", what)
}
func folded(t *testing.T, r *Room, code string) []Message {
t.Helper()
msgs, _, err := r.History(code, 0, 100)
if err != nil {
t.Fatal(err)
}
return Fold(msgs)
}
// An upstream message becomes a channel line attributed to
// `fossil:<xfrom>` with Origin "fossil" (Decision G), and the cursor
// moves to the msgid it consumed.
func TestBridgeInboundMapsAFossilMessage(t *testing.T) {
r, f, store, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 7, XFrom: "drh", XMsg: "the lease is per store"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message to arrive", func() bool { return len(folded(t, r, "codeA")) == 1 })
got := folded(t, r, "codeA")[0]
if got.User != "fossil:drh" || got.Origin != "fossil" || got.Kind != KindMsg {
t.Fatalf("inbound line = %+v", got)
}
if got.Text != "the lease is per store" {
t.Fatalf("inbound text = %q", got.Text)
}
waitFor(t, "the cursor to advance", func() bool {
store.mu.Lock()
defer store.mu.Unlock()
return store.cursor == 7
})
// The poll resumes from the cursor rather than re-reading.
waitFor(t, "a resumed poll", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
for _, p := range f.polls {
if p == "7" {
return true
}
}
return false
})
}
// An mdel entry tombstones the message it names — found through the
// msgid map, because fossil's numbering is not quarry's.
func TestBridgeInboundDeleteBecomesATombstone(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 1, XFrom: "drh", XMsg: "oops"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
f.push(fossilMsg{MsgID: 2, XFrom: "drh", MDel: 1})
waitFor(t, "the tombstone", func() bool {
got := folded(t, r, "codeA")
return len(got) == 1 && got[0].Deleted
})
if got := folded(t, r, "codeA")[0]; got.Text != "" {
t.Fatalf("a tombstoned message kept its text: %+v", got)
}
}
// An attachment is LINKED, never copied: fossil keeps the file, and a
// copy in a seven-day window would be a second, shorter-lived truth.
func TestBridgeInboundLinksAnAttachment(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 3, XFrom: "drh", XMsg: "here it is",
FSize: 1024, FName: "patch.diff"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
text := folded(t, r, "codeA")[0].Text
if !strings.Contains(text, "[patch.diff]("+f.srv.URL+"/chat-download?name=3)") {
t.Fatalf("no attachment link in %q", text)
}
if !strings.Contains(text, "here it is") {
t.Fatalf("the attachment link replaced the message: %q", text)
}
}
// The loop guard: fossil attributes outbound posts to the bridge's own
// login, so its own messages must never come back in.
func TestBridgeInboundIgnoresItsOwnLogin(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(
fossilMsg{MsgID: 4, XFrom: "bridgebot", XMsg: "**dan:** echo of a quarry line"},
fossilMsg{MsgID: 5, XFrom: "drh", XMsg: "a real one"},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the real message", func() bool { return len(folded(t, r, "codeA")) == 1 })
// Give the skipped one every chance to appear before concluding.
time.Sleep(300 * time.Millisecond)
got := folded(t, r, "codeA")
if len(got) != 1 || got[0].Text != "a real one" {
t.Fatalf("the bridge re-imported its own message: %+v", got)
}
}
// With quarry's robot on, the upstream's timeline announcements are
// skipped — otherwise every check-in is announced twice. With it off,
// they are ordinary messages.
func TestBridgeSkipsTheUpstreamTimelineRobotOnlyWhenOursIsOn(t *testing.T) {
announcement := "Add the lease. (check-in: <a href='/info/abc'>abc</a>, user: drh)"
for _, tc := range []struct {
name string
on bool
lines int
}{
{"robot on", true, 1}, {"robot off", false, 2},
} {
t.Run(tc.name, func(t *testing.T) {
r, f, _, b := bridgeFixture(t)
b.RobotOn = func(string) bool { return tc.on }
f.push(
fossilMsg{MsgID: 8, XFrom: "timeline", XMsg: announcement},
fossilMsg{MsgID: 9, XFrom: "drh", XMsg: "a person talking"},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the person's message", func() bool {
for _, m := range folded(t, r, "codeA") {
if m.Text == "a person talking" {
return true
}
}
return false
})
time.Sleep(300 * time.Millisecond)
if got := folded(t, r, "codeA"); len(got) != tc.lines {
t.Fatalf("channel holds %d lines, want %d: %+v", len(got), tc.lines, got)
}
})
}
}
// Outbound: a person's message is posted upstream with the name
// prefix fossil has no other place for. A robot line and a line that
// came FROM fossil are never forwarded.
func TestBridgeOutboundPrefixesAndRefusesEchoes(t *testing.T) {
r, f, _, b := bridgeFixture(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
// Let the tail attach before anything is posted: it starts at NOW,
// so a message published first is correctly not forwarded.
time.Sleep(300 * time.Millisecond)
for _, m := range []Message{
{Repo: "alpha", User: "dan", Text: "hello fossil"},
{Repo: "alpha", User: "robot", Kind: KindRobot, Origin: "robot", Text: "check-in abc"},
{Repo: "alpha", User: "fossil:drh", Origin: "fossil", Text: "came from over there"},
} {
if _, err := r.Post("codeA", m); err != nil {
t.Fatal(err)
}
}
waitFor(t, "the outbound post", func() bool { return len(f.sentMsgs()) > 0 })
time.Sleep(300 * time.Millisecond)
sent := f.sentMsgs()
if len(sent) != 1 {
t.Fatalf("the bridge forwarded %d messages, want 1: %q", len(sent), sent)
}
if sent[0] != "**dan:** hello fossil" {
t.Fatalf("outbound msg = %q", sent[0])
}
}
// A failing upstream is recorded and retried, not spun on: the first
// poll after a 5xx does not happen immediately.
func TestBridgeBacksOffOnFailure(t *testing.T) {
_, f, store, b := bridgeFixture(t)
f.mu.Lock()
f.fail = 3
f.mu.Unlock()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the failure to be recorded", func() bool { return store.err() != "" })
if !strings.Contains(store.err(), "500") {
t.Fatalf("recorded error does not name the status: %q", store.err())
}
// backoffMin is 30 s, so within a second of the first failure there
// must not have been a second poll.
f.mu.Lock()
n := len(f.polls)
f.mu.Unlock()
time.Sleep(time.Second)
f.mu.Lock()
after := len(f.polls)
f.mu.Unlock()
if after > n {
t.Fatalf("the bridge retried %d times inside a second: no backoff", after-n)
}
}
// A refusal fossil delivers inside a 200 JSON envelope is an error,
// not a message: importing it would put "Missing permissions" in the
// channel as if somebody had said it.
func TestBridgeTreatsAnEnvelopedRefusalAsAnError(t *testing.T) {
r, f, store, b := bridgeFixture(t)
f.mu.Lock()
f.noAuth = true
f.mu.Unlock()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the refusal", func() bool { return store.err() != "" })
if !strings.Contains(store.err(), "Missing permissions") {
t.Fatalf("refusal not reported in words: %q", store.err())
}
if strings.Contains(store.err(), "<a href") {
t.Fatalf("the refusal was logged with fossil's markup in it: %q", store.err())
}
if got := folded(t, r, "codeA"); len(got) != 0 {
t.Fatalf("a refusal was imported as a message: %+v", got)
}
}
// A long poll that expires with nothing new answers an empty body.
// That is the quiet case, not a broken upstream: reading it as an
// error would put every idle channel into permanent backoff.
func TestBridgeTreatsAnEmptyBodyAsQuiet(t *testing.T) {
_, f, store, b := bridgeFixture(t)
quiet := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/chat-poll" {
select {
case f.pollHit <- struct{}{}:
default:
}
}
// Nothing at all, exactly as fossil answers an expired poll.
}))
defer quiet.Close()
b.cfg.URL = quiet.URL
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
// Several polls happen inside a couple of seconds, which they
// could not if the first had been recorded as a failure.
for i := 0; i < 2; i++ {
select {
case <-f.pollHit:
case <-time.After(5 * time.Second):
t.Fatal("the bridge stopped polling a quiet upstream")
}
}
if e := store.err(); e != "" {
t.Fatalf("a quiet poll was recorded as a failure: %q", e)
}
}
// The cursor must never move past a message the channel refused. If it
// did, fossil would never offer that message again — it only ever
// hands out what is newer than `name` — and the line would be lost.
func TestBridgeDoesNotAdvancePastAMessageItCouldNotPost(t *testing.T) {
r, f, store, b := bridgeFixture(t)
var refuse atomic.Bool
refuse.Store(true)
b.post = func(code string, m Message) (uint64, error) {
if refuse.Load() {
return 0, errors.New("channel is refusing")
}
return r.Post(code, m)
}
// The retry is the assertion, not the wait: the production floor
// is half a minute and TestBridgeBacksOffOnFailure pins it.
b.BackoffMin, b.BackoffMax = 100*time.Millisecond, 200*time.Millisecond
f.push(
fossilMsg{MsgID: 5, XFrom: "drh", XMsg: "the one that must survive"},
fossilMsg{MsgID: 6, XFrom: "drh", XMsg: "the one after it"},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the refusal to be recorded", func() bool { return store.err() != "" })
// Nothing landed, and — the point — the cursor did not move.
if got := folded(t, r, "codeA"); len(got) != 0 {
t.Fatalf("a refused message reached the channel: %+v", got)
}
store.mu.Lock()
cursor := store.cursor
store.mu.Unlock()
if cursor != 0 {
t.Fatalf("the cursor advanced to %d past a message that was never delivered", cursor)
}
if got := b.cursor(); got != 0 {
t.Fatalf("the in-memory cursor advanced to %d", got)
}
// When the channel recovers, the retry re-reads from the same
// place and the message is not lost.
refuse.Store(false)
f.push(
fossilMsg{MsgID: 5, XFrom: "drh", XMsg: "the one that must survive"},
fossilMsg{MsgID: 6, XFrom: "drh", XMsg: "the one after it"},
)
waitFor(t, "the retry to carry both", func() bool { return len(folded(t, r, "codeA")) == 2 })
got := folded(t, r, "codeA")
if got[0].Text != "the one that must survive" || got[1].Text != "the one after it" {
t.Fatalf("the retry did not replay the batch in order: %+v", got)
}
}
// An entry the msgid map already knows was carried by an earlier poll.
// Posting it again would double it, which is what a replay, a cursor
// reset or a second node briefly running the same bridge produces.
func TestBridgeDoesNotDoublePostAReplayedMessage(t *testing.T) {
r, f, _, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 11, XFrom: "drh", XMsg: "said once"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
// A cursor reset — the honest shape of "this poll is replayed" —
// makes the upstream offer the same msgid again.
b.setCursor(0)
f.push(fossilMsg{MsgID: 11, XFrom: "drh", XMsg: "said once"})
waitFor(t, "the replayed poll to be read", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
n := 0
for _, p := range f.polls {
if p == "0" {
n++
}
}
return n >= 2
})
time.Sleep(300 * time.Millisecond)
if got := folded(t, r, "codeA"); len(got) != 1 {
t.Fatalf("a replayed msgid was posted %d times: %+v", len(got), got)
}
}
// A replayed deletion is recognised as done rather than tombstoning
// twice: the mdel entry's own msgid is mapped like any other.
func TestBridgeDoesNotDoublePostAReplayedDeletion(t *testing.T) {
r, f, store, b := bridgeFixture(t)
f.push(fossilMsg{MsgID: 1, XFrom: "drh", XMsg: "oops"})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
waitFor(t, "the message", func() bool { return len(folded(t, r, "codeA")) == 1 })
f.push(fossilMsg{MsgID: 2, XFrom: "drh", MDel: 1})
waitFor(t, "the tombstone", func() bool {
got := folded(t, r, "codeA")
return len(got) == 1 && got[0].Deleted
})
before, _, err := r.History("codeA", 0, 100)
if err != nil {
t.Fatal(err)
}
if _, ok := store.ChatBridgeSeq("alpha", 2); !ok {
t.Fatal("the deletion's own msgid was not remembered")
}
b.setCursor(0)
f.push(fossilMsg{MsgID: 2, XFrom: "drh", MDel: 1})
waitFor(t, "the replayed deletion to be read", func() bool {
f.mu.Lock()
defer f.mu.Unlock()
n := 0
for _, p := range f.polls {
if p == "0" {
n++
}
}
return n >= 2
})
time.Sleep(300 * time.Millisecond)
after, _, err := r.History("codeA", 0, 100)
if err != nil {
t.Fatal(err)
}
if len(after) != len(before) {
t.Fatalf("a replayed deletion added %d raw messages", len(after)-len(before))
}
}
// A send that fails is a bridge that is not connected: it reaches the
// status line, it backs off, and — the point — the refused message is
// RETRIED rather than dropped. Reading the next Tail message on
// failure would lose the one that failed, silently and for good.
func TestBridgeRecordsAndBacksOffOnOutboundFailure(t *testing.T) {
r, _, store, b := bridgeFixture(t)
var refuse atomic.Bool
refuse.Store(true)
var mu sync.Mutex
var got []string
var attempts atomic.Int64
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
if !strings.HasSuffix(req.URL.Path, "/chat-send") {
return // polls succeed; one direction failing is enough
}
attempts.Add(1)
if refuse.Load() {
http.Error(w, "nope", http.StatusInternalServerError)
return
}
_ = req.ParseForm()
mu.Lock()
got = append(got, req.FormValue("msg"))
mu.Unlock()
}))
defer upstream.Close()
b.cfg.URL = upstream.URL
b.BackoffMin, b.BackoffMax = 100*time.Millisecond, 200*time.Millisecond
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go b.Run(ctx)
time.Sleep(300 * time.Millisecond) // let the tail attach
for _, text := range []string{"first", "second", "third"} {
if _, err := r.Post("codeA", Message{Repo: "alpha", User: "dan", Text: text}); err != nil {
t.Fatal(err)
}
}
waitFor(t, "the outbound failure to be recorded", func() bool { return store.err() != "" })
if !strings.Contains(store.err(), "chat-send") {
t.Fatalf("the recorded error does not name the direction: %q", store.err())
}
// Nothing got through while the upstream was refusing.
mu.Lock()
delivered := len(got)
mu.Unlock()
if delivered != 0 {
t.Fatalf("%d messages were accepted by a refusing upstream", delivered)
}
// Retries are paced: against a 100ms floor, a busy channel cannot
// have produced an unbounded number of attempts.
if n := attempts.Load(); n > 12 {
t.Fatalf("%d send attempts in ~1s against a 100ms backoff floor", n)
}
// On recovery every message arrives, in order, exactly once — the
// first one having been held across the whole outage.
refuse.Store(false)
waitFor(t, "the held messages to be delivered", func() bool {
mu.Lock()
defer mu.Unlock()
return len(got) == 3
})
time.Sleep(400 * time.Millisecond) // any duplicate would land by now
mu.Lock()
defer mu.Unlock()
want := []string{"**dan:** first", "**dan:** second", "**dan:** third"}
if len(got) != len(want) {
t.Fatalf("delivered %d messages, want %d: %q", len(got), len(want), got)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("delivery %d = %q, want %q (all: %q)", i, got[i], want[i], got)
}
}
}
// Redirects are refused rather than followed: Go re-sends the
// Authorization header on a same-host redirect, and this one is a
// fossil password.
func TestBridgeRefusesRedirects(t *testing.T) {
var authSeen atomic.Int64
final := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
if req.Header.Get("Authorization") != "" {
authSeen.Add(1)
}
fmt.Fprint(w, `{"msgs":[]}`)
}))
defer final.Close()
redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
http.Redirect(w, req, final.URL+req.URL.Path, http.StatusFound)
}))
defer redirector.Close()
b := &Bridge{cfg: BridgeConfig{Repo: "alpha", URL: redirector.URL,
User: "bridgebot", Secret: "s3cret"}, Client: bridgeClient(5*time.Second, true)}
_, err := b.get(context.Background(), redirector.URL+"/chat-poll?name=0&raw=1")
if err == nil {
t.Fatal("a redirect was followed")
}
if !strings.Contains(err.Error(), "redirect") {
t.Fatalf("the refusal does not say why: %v", err)
}
if n := authSeen.Load(); n != 0 {
t.Fatalf("the credential was carried to the redirect target %d times", n)
}
}
// A server-side fetcher holding a credential does not reach loopback,
// private or link-local addresses unless an operator said so — the
// metadata service lives at a link-local address.
func TestBridgeRefusesPrivateAddressesUnlessAllowed(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
fmt.Fprint(w, `{"msgs":[]}`)
}))
defer srv.Close()
guarded := &Bridge{cfg: BridgeConfig{Repo: "alpha", URL: srv.URL, User: "u", Secret: "p"},
Client: bridgeClient(5*time.Second, false)}
if _, err := guarded.get(context.Background(), srv.URL+"/chat-poll"); err == nil {
t.Fatal("a loopback upstream was reached with the guard on")
} else if !strings.Contains(err.Error(), "allow-private-bridges") {
t.Fatalf("the refusal does not say how to permit it: %v", err)
}
allowed := &Bridge{cfg: BridgeConfig{Repo: "alpha", URL: srv.URL, User: "u", Secret: "p"},
Client: bridgeClient(5*time.Second, true)}
if _, err := allowed.get(context.Background(), srv.URL+"/chat-poll"); err != nil {
t.Fatalf("the opt-in did not permit a loopback upstream: %v", err)
}
for _, addr := range []string{"127.0.0.1", "169.254.169.254", "10.0.0.5", "192.168.1.1", "::1"} {
if !reserved(net.ParseIP(addr)) {
t.Errorf("%s is not treated as reserved", addr)
}
}
for _, addr := range []string{"93.184.216.34", "8.8.8.8"} {
if reserved(net.ParseIP(addr)) {
t.Errorf("%s is treated as reserved", addr)
}
}
}
// Credentials and queries never reach an error string, because error
// strings are copied into the status row an admin page renders.
func TestSafeURLDropsCredentialsAndQuery(t *testing.T) {
for raw, want := range map[string]string{
"https://bot:[email protected]/repo/chat-poll?name=3": "https://fossil.example/repo/chat-poll",
"https://fossil.example/repo?token=abc": "https://fossil.example/repo",
"://nonsense": "the upstream",
} {
if got := safeURL(raw); got != want {
t.Errorf("safeURL(%q) = %q, want %q", raw, got, want)
}
}
}
// An upstream filename is the upstream's, so it cannot be allowed to
// close the link markdown puts it in.
func TestAttachmentFilenameIsEscaped(t *testing.T) {
b := &Bridge{cfg: BridgeConfig{URL: "https://fossil.example/r"}}
text := b.textOf(fossilMsg{MsgID: 4, FSize: 12, FName: "eviL](javascript:alert(1)).txt"})
// The bracket must arrive escaped: unescaped, it would close the
// label and make the rest of the filename the link's target.
if strings.Contains(text, `L](javascript:`) {
t.Fatalf("a filename closed its own link: %q", text)
}
if !strings.Contains(text, `L\](javascript:`) {
t.Fatalf("the bracket was not escaped: %q", text)
}
if !strings.Contains(text, "https://fossil.example/r/chat-download?name=4") {
t.Fatalf("the link lost its target: %q", text)
}
}
// The address that was checked must be the address that is dialed.
// Re-resolving the hostname after the check is the DNS-rebinding gap:
// the second answer need not be the first, and a guard that leaves it
// open is a guard that only inconveniences an attacker.
func TestBridgeDialsTheAddressItValidated(t *testing.T) {
var hits atomic.Int64
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
hits.Add(1)
fmt.Fprint(w, `{"msgs":[]}`)
}))
defer srv.Close()
_, port, err := net.SplitHostPort(strings.TrimPrefix(srv.URL, "http://"))
if err != nil {
t.Fatal(err)
}
// A resolver that passes the check with a public address and then
// hands back loopback on every later lookup. A dialer that
// re-resolves reaches the test server; one that dials what it
// checked goes to the public address and never arrives.
var lookups atomic.Int64
restore := lookupIP
t.Cleanup(func() { lookupIP = restore })
lookupIP = func(_ context.Context, _ string) ([]net.IPAddr, error) {
if lookups.Add(1) == 1 {
return []net.IPAddr{{IP: net.ParseIP("203.0.113.1")}}, nil // TEST-NET-3
}
return []net.IPAddr{{IP: net.ParseIP("127.0.0.1")}}, nil
}
b := &Bridge{cfg: BridgeConfig{Repo: "alpha", URL: "http://fossil.example:" + port,
User: "u", Secret: "p"}, Client: bridgeClient(2*time.Second, false)}
_, err = b.get(context.Background(), "http://fossil.example:"+port+"/chat-poll")
if err == nil {
t.Fatal("the bridge connected to the rebound address")
}
// The check ran against the FIRST answer, which passed — so the
// failure must not be the private-address refusal.
if strings.Contains(err.Error(), "private address") {
t.Fatalf("the guard validated the second answer, not the first: %v", err)
}
// And this is the assertion that matters: the loopback server the
// second answer pointed at was never reached.
if n := hits.Load(); n != 0 {
t.Fatalf("the rebound loopback address was dialed %d times", n)
}
if lookups.Load() == 0 {
t.Fatal("the guard never resolved the host")
}
}