// bridge.go is RFC 0011 §9: an opt-in connector between one quarry
// channel and one Fossil server's chat.
//
// Fossil's chat is a server-local table, never synced, gated by `C`,
// long-polled through /chat-poll and written through /chat-send. Every
// key below was checked against a real fossil 2.28 (`msgid`, `mtime`,
// `xfrom`, `xmsg`, `fsize`, `fname`, `mdel`) rather than assumed.
//
// Two constraints that a reading of chat.c alone does not surface, and
// that a bridge silently fails without:
//
// 1. HTTP Basic authentication is refused unless the upstream has the
// `http_authentication_ok` config entry set. Without it every
// request is anonymous and every answer is a permissions error.
// 2. /chat-poll creates its full-text TEMP triggers on the way in, so
// fossil's CSRF gate must consider the request safe or the poll
// dies with "not authorized: CREATE TEMP TRIGGER". A same-origin
// `Referer` header is what makes it safe, so every request here
// carries one.
//
// What the bridge does not pretend: fossil messages cannot be edited
// from quarry, two fossil servers are two chats, and a bridged
// channel's history is still quarry's seven-day window.
package chat
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"net/url"
"regexp"
"strings"
"sync"
"time"
)
const (
// pollTimeout must exceed fossil's chat-poll-timeout (420 s by
// default) or every poll is a client-side cancellation that reads
// nothing and looks, from the log, like an upstream that is down.
pollTimeout = 9 * time.Minute
// sendTimeout is an ordinary write; nothing about it hangs.
sendTimeout = 30 * time.Second
// backoffMin/backoffMax pace retries after a failed poll. Doubling
// from half a minute to five keeps a broken upstream from being
// hammered while a transient one recovers within a minute.
backoffMin = 30 * time.Second
backoffMax = 5 * time.Minute
)
// BridgeConfig is one bridged channel: which repository, which channel
// key, and the upstream credential the poller replays.
type BridgeConfig struct {
Repo string
Code string
URL string
User string
Secret string
// LastMsgID is where the poller resumes; 0 asks fossil for the
// channel's recent history rather than for nothing.
LastMsgID int64
// AllowPrivate permits an upstream on a loopback, private or
// link-local address. Off by default: the bridge is a server-side
// fetcher holding a credential, so an admin who can name a URL
// could otherwise point it at the metadata service or at anything
// else inside the network the forge sits in. Operators bridging a
// fossil on the same host turn it on deliberately
// (--allow-private-bridges), and the tests set it because their
// fossil is on 127.0.0.1.
AllowPrivate bool
}
// BridgeStore is the cursor and the msgid map. Both are OLTP, both are
// window-bounded, and a nil store is a bridge with no memory: it still
// carries messages, it just re-reads the upstream from where it
// started every time the process restarts.
type BridgeStore interface {
// SetChatBridgeCursor advances how far the poller has read.
SetChatBridgeCursor(repo string, msgid int64, at time.Time) error
// RecordChatBridgeError remembers why the last poll failed.
RecordChatBridgeError(repo string, err error) error
// MapChatBridgeMsg remembers which sequence an upstream msgid
// became, so a later mdel can tombstone the right message.
MapChatBridgeMsg(repo string, msgid int64, seq uint64) error
// ChatBridgeSeq answers that question back.
ChatBridgeSeq(repo string, msgid int64) (uint64, bool)
}
// Bridge carries one channel both ways.
type Bridge struct {
room *Room
store BridgeStore
cfg BridgeConfig
// Client, SendClient, BackoffMin and BackoffMax are SEAMS, not
// configuration: NewBridge fills all four with the values §9 calls
// for, and nothing in the composition root sets them. They are
// exported so a test in another package could drive a bridge
// against a stub, and so this package's own tests can shorten a
// half-minute backoff to a hundred milliseconds without waiting it
// out. Changing them in production changes the protocol timings.
//
// Client is the poller's; its timeout must exceed the upstream's
// poll hang. SendClient is the short-timeout one for outbound.
Client *http.Client
SendClient *http.Client
// RobotOn reports whether this repository's own activity robot is
// on. When it is, the upstream's timeline announcements are
// skipped: otherwise every check-in is announced twice, once by
// each robot, which is the failure that teaches people to mute the
// channel. nil means "off", the direction this must fail in.
RobotOn func(repo string) bool
// timelineUser is the upstream's chat-timeline-user when it could
// be learned; "" falls back to the shape heuristic below.
timelineUser string
// BackoffMin and BackoffMax pace retries in BOTH directions.
BackoffMin, BackoffMax time.Duration
// mu guards the cursor alone. The inbound loop advances it while
// anything else — a test, a future status call — may read it.
mu sync.Mutex
// post is Room.Post, named so a test can make the channel refuse.
// "The cursor does not move past a message that was not
// delivered" is the property this bridge lives or dies by, and it
// cannot be proved against a broker that always accepts.
post func(code string, m Message) (uint64, error)
}
func NewBridge(room *Room, store BridgeStore, cfg BridgeConfig) *Bridge {
b := &Bridge{
room: room, store: store, cfg: cfg,
Client: bridgeClient(pollTimeout, cfg.AllowPrivate),
SendClient: bridgeClient(sendTimeout, cfg.AllowPrivate),
}
b.BackoffMin, b.BackoffMax = backoffMin, backoffMax
b.post = func(code string, m Message) (uint64, error) { return b.room.Post(code, m) }
return b
}
// bridgeClient is the only kind of HTTP client this package makes.
//
// Redirects are REFUSED rather than followed. Go re-sends an
// Authorization header on a same-host redirect, and the bridge's
// header is a fossil password: a compromised or merely misconfigured
// upstream that answers 302 could otherwise walk the credential
// somewhere the operator never named. A bridge that needs a redirect
// is a bridge configured with the wrong URL, and saying so is better
// than quietly following one.
//
// The dialer refuses private destinations unless the operator allowed
// them. It is enforced at CONNECT time, on the resolved address, so a
// public name that resolves inward — DNS rebinding, or an internal
// split-horizon record — is refused too; a check on the URL string
// alone would not see it.
//
// The address that was CHECKED is the address that is DIALED. Handing
// the hostname back to the dialer would re-resolve it, and the second
// answer need not be the first: that gap is the whole DNS-rebinding
// attack, and closing it is the only reason to do this in a dialer
// rather than on the URL. TLS is unaffected — the transport runs the
// handshake itself, after this, using the hostname from the request,
// so SNI and certificate verification still name the host.
func bridgeClient(timeout time.Duration, allowPrivate bool) *http.Client {
tr := http.DefaultTransport.(*http.Transport).Clone()
if !allowPrivate {
d := &net.Dialer{Timeout: 15 * time.Second, KeepAlive: 30 * time.Second}
tr.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) {
host, port, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
ips, err := lookupIP(ctx, host)
if err != nil {
return nil, err
}
if len(ips) == 0 {
return nil, fmt.Errorf("no address for %s", host)
}
for _, ip := range ips {
if reserved(ip.IP) {
return nil, fmt.Errorf(
"refusing a chat bridge to a private address (%s); "+
"start quarry with --allow-private-bridges to permit it", ip.IP)
}
}
// The first address, because every one of them passed.
return d.DialContext(ctx, network, net.JoinHostPort(ips[0].IP.String(), port))
}
}
return &http.Client{
Timeout: timeout,
Transport: tr,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return fmt.Errorf("chat bridge: %s redirected to %s; configure the final URL",
safeURL(via[len(via)-1].URL.String()), safeURL(req.URL.String()))
},
}
}
// lookupIP is the resolver the guard uses, named so a test can hand it
// a resolver that answers differently the second time — which is the
// only way to prove the guard dials what it checked.
var lookupIP = func(ctx context.Context, host string) ([]net.IPAddr, error) {
return net.DefaultResolver.LookupIPAddr(ctx, host)
}
// reserved reports whether an address is one a server-side fetcher has
// no business reaching on its own initiative: loopback, link-local
// (which covers the 169.254.169.254 metadata service), private ranges,
// unspecified and multicast.
func reserved(ip net.IP) bool {
return ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() ||
ip.IsPrivate() || ip.IsUnspecified() || ip.IsMulticast() ||
ip.IsInterfaceLocalMulticast()
}
// safeURL is a URL as it may appear in an error, a log line or a
// stored status: scheme, host and path, with any userinfo and query
// removed. Credentials and tokens live in exactly those two places,
// and an error string is copied into the OLTP row a status page
// renders.
func safeURL(raw string) string {
u, err := url.Parse(raw)
if err != nil || u.Host == "" {
return "the upstream"
}
return u.Scheme + "://" + u.Host + u.Path
}
// fossilMsg is one /chat-poll entry, field for field as fossil emits
// it. Anything not named here is fossil's presentation (uclr, lmtime)
// and has no meaning on this side.
type fossilMsg struct {
MsgID int64 `json:"msgid"`
MTime string `json:"mtime"`
XFrom string `json:"xfrom"`
XMsg string `json:"xmsg"`
FSize int64 `json:"fsize"`
FName string `json:"fname"`
MDel int64 `json:"mdel"`
IsError bool `json:"isError"`
}
type fossilPoll struct {
Msgs []fossilMsg `json:"msgs"`
}
// Run carries the channel both ways until ctx is done. Inbound is a
// long poll; outbound is a tail of the channel from NOW — a bridge
// starting up must not replay the window into the upstream, which
// would post a week of quarry's conversation as new fossil messages.
func (b *Bridge) Run(ctx context.Context) {
if b == nil || b.room == nil || b.cfg.URL == "" {
return
}
b.timelineUser = b.fetchTimelineUser(ctx)
go b.outbound(ctx)
b.inbound(ctx)
}
func (b *Bridge) inbound(ctx context.Context) {
backoff := b.BackoffMin
for ctx.Err() == nil {
n, err := b.pollOnce(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
log.Printf("chat: bridge %s: poll: %v", b.cfg.Repo, err)
b.fail(err)
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff *= 2; backoff > b.BackoffMax {
backoff = b.BackoffMax
}
continue
}
backoff = b.BackoffMin
if n == 0 {
// A poll that hung out its timeout and returned nothing is
// the ordinary quiet case, not a reason to pause — but a
// server answering instantly with nothing would spin, so
// the loop yields briefly either way.
select {
case <-ctx.Done():
return
case <-time.After(time.Second):
}
}
}
}
// pollOnce reads one /chat-poll response into the channel and answers
// how many upstream entries it consumed.
func (b *Bridge) pollOnce(ctx context.Context) (int, error) {
poll := fmt.Sprintf("%s/chat-poll?name=%d&raw=1", strings.TrimRight(b.cfg.URL, "/"), b.cursor())
body, err := b.get(ctx, poll)
if err != nil {
return 0, err
}
// A poll that hangs out its timeout with nothing new answers an
// EMPTY BODY, not an empty envelope (verified against fossil 2.28).
// Reading that as malformed JSON would turn every quiet interval
// into a recorded failure and a thirty-second backoff — a bridge
// that works only on a busy channel.
if len(strings.TrimSpace(string(body))) == 0 {
return 0, nil
}
var out fossilPoll
if err := json.Unmarshal(body, &out); err != nil {
// A fossil that refuses answers HTML, not JSON. Saying so is
// more use than "invalid character '<'".
return 0, fmt.Errorf("chat-poll did not answer JSON (upstream refused or is not a fossil server): %w", err)
}
// The cursor advances one entry at a time, and only PAST an entry
// this bridge is finished with. A batch that fails halfway leaves
// the cursor on the last entry that landed, so the retry after the
// backoff re-reads from the message that did not — the alternative
// drops it for ever, since fossil will never offer it again.
n := 0
for _, m := range out.Msgs {
if m.IsError {
return n, errors.New("chat-poll: " + plainText(m.XMsg))
}
if err := b.consume(m); err != nil {
return n, err
}
b.advance(m.MsgID)
n++
}
return n, nil
}
// consume turns one upstream entry into a channel line.
//
// nil means "this entry is finished with" — carried, or deliberately
// skipped — and only then may the caller advance the cursor past it.
// An error means the entry was NOT delivered: the caller stops the
// batch there, so the retry after the backoff starts at this same
// msgid. Advancing regardless would drop the message permanently,
// because fossil only ever offers what is newer than `name`.
//
// An entry the msgid map already knows was carried by an earlier poll
// — a replay, a cursor reset, or a second node briefly running the
// same bridge — and posting it again would double it in the channel.
func (b *Bridge) consume(m fossilMsg) error {
if b.alreadyBridged(m.MsgID) {
return nil
}
switch {
case m.MDel != 0:
// A deletion names the msgid it removed. Without a mapping the
// message is either one the bridge never imported or one the
// window has already dropped, and in both cases there is
// nothing here to tombstone.
if b.store == nil {
return nil
}
seq, ok := b.store.ChatBridgeSeq(b.cfg.Repo, m.MDel)
if !ok {
return nil
}
tomb, err := b.post(b.cfg.Code, Message{
Repo: b.cfg.Repo, Kind: KindDel, Ref: seq,
Origin: "fossil", User: bridgeUser(m.XFrom),
})
if err != nil {
return fmt.Errorf("tombstone for #%d: %w", seq, err)
}
// The tombstone's OWN msgid is mapped too, so a replayed
// deletion is recognised as done rather than posted twice.
b.remember(m.MsgID, tomb)
return nil
case b.skip(m):
return nil
default:
text := b.textOf(m)
if strings.TrimSpace(text) == "" {
return nil
}
seq, err := b.post(b.cfg.Code, Message{
Repo: b.cfg.Repo, Kind: KindMsg, Origin: "fossil",
User: bridgeUser(m.XFrom), Text: text,
})
if err != nil {
return fmt.Errorf("importing msgid %d: %w", m.MsgID, err)
}
b.remember(m.MsgID, seq)
return nil
}
}
// alreadyBridged reports whether this upstream msgid is already in the
// channel. Without a store there is no memory and every poll is taken
// at face value, which is the documented cost of running with none.
func (b *Bridge) alreadyBridged(msgid int64) bool {
if b.store == nil || msgid == 0 {
return false
}
_, ok := b.store.ChatBridgeSeq(b.cfg.Repo, msgid)
return ok
}
func (b *Bridge) remember(msgid int64, seq uint64) {
if b.store == nil {
return
}
if err := b.store.MapChatBridgeMsg(b.cfg.Repo, msgid, seq); err != nil {
log.Printf("chat: bridge %s: mapping msgid %d: %v", b.cfg.Repo, msgid, err)
}
}
// cursor and setCursor are the only readers and writers of
// cfg.LastMsgID once Run has started. The poll loop owns it in
// practice, but it is read from outside — so it is taken under the
// lock rather than left as a field two goroutines happen to touch.
func (b *Bridge) cursor() int64 {
b.mu.Lock()
defer b.mu.Unlock()
return b.cfg.LastMsgID
}
func (b *Bridge) setCursor(msgid int64) {
b.mu.Lock()
defer b.mu.Unlock()
b.cfg.LastMsgID = msgid
}
func (b *Bridge) advance(msgid int64) {
if msgid <= b.cursor() {
return
}
b.setCursor(msgid)
if b.store != nil {
if err := b.store.SetChatBridgeCursor(b.cfg.Repo, msgid, time.Now()); err != nil {
log.Printf("chat: bridge %s: cursor: %v", b.cfg.Repo, err)
}
}
}
// skip is the two reasons an inbound message is not carried.
//
// The loop guard is first and unconditional: outbound posts arrive at
// the upstream under the bridge's own login, so re-importing them
// would echo every quarry message back into quarry.
//
// The second is the upstream's timeline robot, skipped only while
// quarry's own robot is announcing the same events (§10). Fossil does
// not expose `chat-timeline-user` on any endpoint a bridge login can
// read, so the name is used when /json/config volunteered it and the
// announcement's own SHAPE is used when it did not: fossil builds
// those messages with a trailing "(check-in: <a href=…" clause that no
// hand-typed message carries.
func (b *Bridge) skip(m fossilMsg) bool {
if m.XFrom == "" || strings.EqualFold(m.XFrom, b.cfg.User) {
return true
}
robotOn := b.RobotOn != nil && b.RobotOn(b.cfg.Repo)
if !robotOn {
return false
}
if b.timelineUser != "" {
return strings.EqualFold(m.XFrom, b.timelineUser)
}
return timelineShape.MatchString(m.XMsg)
}
// timelineShape is fossil's chat_msg_from_event() format: the event's
// own words followed by a parenthesised artifact clause it builds as
// HTML even in a raw message.
var timelineShape = regexp.MustCompile(`\((check-in|artifact|ticket|wiki|forum post|event): <a href=`)
// textOf is the message body as the channel should hold it: fossil's
// markdown verbatim, plus a link to any attachment. Fossil keeps the
// file itself; the bridge links it rather than copying it, because a
// copy in the channel window would be a second, shorter-lived truth.
func (b *Bridge) textOf(m fossilMsg) string {
text := m.XMsg
if m.FSize <= 0 {
return text
}
name := m.FName
if name == "" {
name = "attachment"
}
// The name is the upstream's, so it is escaped rather than
// trusted: a filename holding `](javascript:...)` would otherwise
// close this link and open one the uploader chose.
link := fmt.Sprintf("[%s](%s/chat-download?name=%d) (%d bytes)",
escapeLinkText(name), strings.TrimRight(b.cfg.URL, "/"), m.MsgID, m.FSize)
if strings.TrimSpace(text) == "" {
return link
}
return text + "\n\n" + link
}
// bridgeUser is Decision G: a bridged line shows as `fossil:<xfrom>`.
// There is no account mapping — the prefix is the honest statement
// that this name belongs to another server's user table, not to a
// quarry account somebody could be confused for.
func bridgeUser(xfrom string) string {
if xfrom == "" {
return "fossil:anonymous"
}
return "fossil:" + xfrom
}
// outbound posts quarry's own messages upstream. Only a person's:
// robot lines are quarry's announcements about quarry, and a
// fossil-origin line came FROM there — echoing either back is the loop
// this half has to not be.
func (b *Bridge) outbound(ctx context.Context) {
msgs, cancel := b.room.Tail(b.cfg.Code, 0)
defer cancel()
// A send that fails is a bridge that is not connected, so it is
// recorded where the status line reads from — an upstream
// accepting polls while refusing sends would otherwise show as
// healthy while half of it was down.
//
// A message is HELD until the upstream takes it. The next Tail
// message is not read while one is pending, so a refused message
// is retried rather than skipped, order is preserved, and the rest
// queue at the broker — Tail never drops, so nothing is lost while
// this loop waits.
backoff := b.BackoffMin
var lastLog time.Time
var pending *Message
for {
if pending == nil {
select {
case <-ctx.Done():
return
case m, ok := <-msgs:
if !ok {
return
}
if m.Origin != "" || m.Kind != KindMsg || strings.TrimSpace(m.Text) == "" {
continue
}
held := m
pending = &held
}
}
err := b.send(ctx, *pending)
if err == nil {
pending = nil
backoff = b.BackoffMin
continue
}
if ctx.Err() != nil {
return
}
b.fail(err)
// One line per backoff window. A channel that is busy while
// the upstream is down would otherwise write a line per
// attempt, which buries every other bridge in the log.
if time.Since(lastLog) >= backoff {
log.Printf("chat: bridge %s: sending #%d: %v", b.cfg.Repo, pending.Seq, err)
lastLog = time.Now()
}
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff *= 2; backoff > b.BackoffMax {
backoff = b.BackoffMax
}
}
}
// fail records why the bridge is not working, for the status line.
// Both directions report here: "connected" has to mean both.
func (b *Bridge) fail(err error) {
if b.store == nil || err == nil {
return
}
if serr := b.store.RecordChatBridgeError(b.cfg.Repo, err); serr != nil {
log.Printf("chat: bridge %s: recording the failure: %v", b.cfg.Repo, serr)
}
}
// send posts one quarry message to /chat-send. Fossil attributes
// everything to the login that sent it, so the author's name is
// carried in the text — the only place the protocol leaves for it.
func (b *Bridge) send(ctx context.Context, m Message) error {
user := m.User
if user == "" {
user = "anonymous"
}
form := url.Values{"msg": {"**" + user + ":** " + m.Text}}
req, err := http.NewRequestWithContext(ctx, "POST",
strings.TrimRight(b.cfg.URL, "/")+"/chat-send", strings.NewReader(form.Encode()))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
b.arm(req)
resp, err := b.SendClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("chat-send to %s: %s", safeURL(b.cfg.URL), resp.Status)
}
// A refusal is a 200 carrying a JSON error object; success is an
// empty body. Reading the status alone would call every rejected
// message delivered.
if e := errorIn(body); e != "" {
return errors.New("chat-send: " + e)
}
return nil
}
func (b *Bridge) get(ctx context.Context, u string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
if err != nil {
return nil, err
}
b.arm(req)
resp, err := b.Client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("%s: %s", safeURL(u), resp.Status)
}
return body, nil
}
// arm puts the credential and the same-origin Referer on a request.
// The Referer is not decoration: fossil treats a request without one
// as CSRF-unsafe and refuses /chat-poll's own table setup, which
// surfaces as a database error rather than as an auth failure.
func (b *Bridge) arm(req *http.Request) {
if b.cfg.User != "" {
req.SetBasicAuth(b.cfg.User, b.cfg.Secret)
}
req.Header.Set("Referer", strings.TrimRight(b.cfg.URL, "/")+"/chat")
}
// fetchTimelineUser asks the upstream for its chat-timeline-user once,
// at startup. Most fossil builds ship without the JSON API and answer
// HTML or 404; that is expected, costs one request, and leaves the
// shape heuristic in charge.
func (b *Bridge) fetchTimelineUser(ctx context.Context) string {
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
body, err := b.get(ctx, strings.TrimRight(b.cfg.URL, "/")+"/json/config/chat-timeline-user")
if err != nil {
return ""
}
var out struct {
Payload map[string]any `json:"payload"`
}
if json.Unmarshal(body, &out) != nil {
return ""
}
name, _ := out.Payload["chat-timeline-user"].(string)
return name
}
// errorIn answers the message inside a fossil JSON error object, "" if
// the body is not one.
func errorIn(body []byte) string {
var e struct {
IsError bool `json:"isError"`
XMsg string `json:"xmsg"`
}
if json.Unmarshal(body, &e) != nil || !e.IsError {
return ""
}
return plainText(e.XMsg)
}
// plainText strips the anchor markup fossil puts in its own error
// strings, so a log line reads as a sentence.
func plainText(s string) string {
var b strings.Builder
depth := 0
for _, r := range s {
switch {
case r == '<':
depth++
case r == '>' && depth > 0:
depth--
case depth == 0:
b.WriteRune(r)
}
}
return strings.TrimSpace(b.String())
}
// escapeLinkText makes an upstream-supplied string safe as the text of
// a markdown link: the brackets that would end it, and the backslash
// that would smuggle one back in, are escaped. Newlines go too — a
// link label cannot span lines, and one that tries ends the link.
func escapeLinkText(s string) string {
r := strings.NewReplacer(
`\`, `\\`, `[`, `\[`, `]`, `\]`, "\n", " ", "\r", " ")
return r.Replace(s)
}