Blacktip/blacktip.go

// Package blacktip is the edge: it owns public IPs, opens the ports hammerhead
// asks for, and forwards each connection to a site (or to its own machine).
package blacktip

import (
	"bytes"
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/http/httputil"
	"net/netip"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.veillax.com/Veillax/reef/docker"
	"git.veillax.com/Veillax/reef/model"
	"git.veillax.com/Veillax/reef/tunnel"
	"git.veillax.com/Veillax/reef/wg"
)

type Config struct {
	Hammerhead string // base URL of the control plane
	Token      string
	TunnelAddr string // UDP listen address for the WireGuard tunnel, e.g. ":51820"
	DataDir    string
	Docker     bool // report local Docker containers for discovery

	// Version is the release this binary was built from, reported to Hammerhead. AutoUpdate says
	// whether the program may update itself; only then does it act on an update request from
	// Hammerhead, by calling UpdateNow (which normally restarts the process into the new binary).
	Version    string
	AutoUpdate bool
	UpdateNow  func(ctx context.Context, progress func(stage string, pct int, line string)) (newVersion string, err error)

	// Traefik, when its Path is set, serves http/https resources instead of the built-in proxy.
	Traefik TraefikConfig

	// Test hooks for mail egress: resolve mail servers without DNS, allow loopback
	// destinations, and choose what the "port 25 open" probe connects to.
	MailLookup       func(domain string) ([]MailHost, error)
	MailAllowPrivate bool
	MailProbe        string
	MailLookupTXT    func(name string) ([]string, error)     // DNS TXT lookups (DKIM check, MTA-STS record)
	MailSTS          func(domain string) (*STSPolicy, error) // replaces MTA-STS discovery
	MailRootCAs      *x509.CertPool                          // trust anchors for verified TLS to mail servers

	// GetCertificate replaces Let's Encrypt for https ports (used by tests).
	GetCertificate func(*tls.ClientHelloInfo) (*tls.Certificate, error)
}

// mapState is the runtime half of a mapping: health, counters and the web proxy.
type mapState struct {
	b      *Blacktip
	m      model.Mapping
	sig    string
	cancel context.CancelFunc

	mu     sync.Mutex
	health []model.TargetHealth
	rr     atomic.Uint64

	conns, bytesIn, bytesOut atomic.Int64
	epoch                    atomic.Int64 // current login generation; kept out of sig so a bump doesn't rebuild the resource

	relayLn   net.Listener // traefik mode: loopback port that reaches the targets through the tunnel
	relayPort int

	transport *http.Transport
	proxy     *httputil.ReverseProxy
}

type Blacktip struct {
	cfg   Config
	wgKey string
	wgPub string
	http  *http.Client
	ctx   context.Context

	mu         sync.Mutex
	version    int64
	last       *model.EdgeConfig
	tun        *wg.Device
	tunIP      string
	peerIP     map[string]string // site ID -> tunnel address
	mail       map[string]model.MailPolicy
	mailSig    string
	mailChecks map[string]model.MailCheck
	mailMu     sync.Mutex
	mailRate   map[string][]time.Time
	sts        stsCache
	local      map[string]bool
	authKey    string
	dashboard  string
	maps       map[string]*mapState
	listeners  map[string]*listener
	status     map[string]string // mapping ID -> listener error

	acme acmeState
	tr   *traefikMgr

	logq       chan model.LogEntry
	reportMu   sync.Mutex
	updating   atomic.Bool
	addrSig    string // machine addresses a wildcard website was spread across, "" when none was
	updateMsg  string // outcome of the update Hammerhead asked for, reported back
	updateID   string // the request that updateMsg answers
	updStage   string // progress of the running update, reported to Hammerhead
	updPct     int
	updLog     []string
	updSent    time.Time
	dockerMu   sync.Mutex
	dockerAt   time.Time
	dockerList []model.Container
}

func New(cfg Config) (*Blacktip, error) {
	b := &Blacktip{
		cfg:        cfg,
		http:       &http.Client{Timeout: 40 * time.Second},
		maps:       map[string]*mapState{},
		listeners:  map[string]*listener{},
		status:     map[string]string{},
		peerIP:     map[string]string{},
		mail:       map[string]model.MailPolicy{},
		mailChecks: map[string]model.MailCheck{},
		mailRate:   map[string][]time.Time{},
		local:      map[string]bool{},
		logq:       make(chan model.LogEntry, 5000),
		ctx:        context.Background(),
	}
	key, err := loadOrCreateKey(filepath.Join(cfg.DataDir, "wg.key"))
	if err != nil {
		return nil, err
	}
	b.wgKey = key
	if b.wgPub, err = wg.PublicKey(key); err != nil {
		return nil, err
	}
	if cfg.Traefik.Path != "" {
		if b.tr, err = newTraefik(b, cfg.Traefik); err != nil {
			return nil, err
		}
	}
	return b, nil
}

// loadOrCreateKey keeps this machine's WireGuard identity across restarts.
func loadOrCreateKey(path string) (string, error) {
	if b, err := os.ReadFile(path); err == nil {
		if k := strings.TrimSpace(string(b)); k != "" {
			return k, nil
		}
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		return "", err
	}
	k, err := wg.GenerateKey()
	if err != nil {
		return "", err
	}
	return k, os.WriteFile(path, []byte(k+"\n"), 0o600)
}

// tunnelPort extracts the UDP port from the -tunnel listen address.
func tunnelPort(addr string) int {
	_, p, err := net.SplitHostPort(addr)
	if err != nil {
		return 0
	}
	n, _ := strconv.Atoi(p)
	return n
}

// Run blocks until ctx is cancelled.
func (b *Blacktip) Run(ctx context.Context) error {
	b.ctx = ctx
	log.Printf("blacktip started; WireGuard public key %s, UDP %s", b.wgPub, b.cfg.TunnelAddr)
	if b.tr != nil {
		go b.tr.run(ctx)
		log.Printf("serving http/https with Traefik (%s)", b.cfg.Traefik.Path)
	}
	go b.reportLoop(ctx)
	go b.mailCheckLoop(ctx)
	go b.retryLoop(ctx)
	go b.shipLogs(ctx)

	for ctx.Err() == nil {
		cfg, err := b.fetchConfig(ctx)
		if err != nil {
			log.Printf("config: %v", err)
			select {
			case <-time.After(3 * time.Second):
			case <-ctx.Done():
			}
			continue
		}
		if cfg == nil {
			continue // long-poll timed out, nothing changed
		}
		b.apply(cfg)
	}
	b.mu.Lock()
	for _, l := range b.listeners {
		l.close()
	}
	for _, ms := range b.maps {
		ms.stop()
	}
	if b.tun != nil {
		b.tun.Close()
	}
	b.mu.Unlock()
	return nil
}

func (b *Blacktip) fetchConfig(ctx context.Context) (*model.EdgeConfig, error) {
	b.mu.Lock()
	v := b.version
	b.mu.Unlock()
	req, _ := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/edge/config?v=%d", b.cfg.Hammerhead, v), nil)
	req.Header.Set("Authorization", "Bearer "+b.cfg.Token)
	resp, err := b.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNoContent {
		return nil, nil
	}
	if resp.StatusCode != 200 {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
		return nil, fmt.Errorf("hammerhead said %d: %s", resp.StatusCode, bytes.TrimSpace(body))
	}
	var cfg model.EdgeConfig
	return &cfg, json.NewDecoder(resp.Body).Decode(&cfg)
}

// hostIPv4 lists every IPv4 address on this machine, loopback included: what a wildcard bind covers.
func hostIPv4() []string {
	var ips []string
	addrs, _ := net.InterfaceAddrs()
	for _, a := range addrs {
		if ipn, ok := a.(*net.IPNet); ok && ipn.IP.To4() != nil {
			ips = append(ips, ipn.IP.String())
		}
	}
	sort.Strings(ips)
	return ips
}

// bindIPs says which addresses a mapping listens on. Normally that is its own BindIP. A website on
// 0.0.0.0 steps aside for websites of the same kind that name an address on the same port: it then
// listens on the machine's other addresses only, so a wildcard never takes a port from a specific IP.
func bindIPs(m model.Mapping, all []model.Mapping) []string {
	if !model.Web(m.Proto) {
		return []string{m.BindIP}
	}
	if ip := net.ParseIP(m.BindIP); ip == nil || !ip.IsUnspecified() {
		return []string{m.BindIP}
	}
	named := map[string]bool{}
	for _, o := range all {
		if o.ID == m.ID || o.Proto != m.Proto || o.Port != m.Port {
			continue
		}
		if ip := net.ParseIP(o.BindIP); ip != nil && !ip.IsUnspecified() {
			named[ip.String()] = true
		}
	}
	if len(named) == 0 {
		return []string{m.BindIP}
	}
	var out []string
	for _, ip := range hostIPv4() {
		if !named[ip] {
			out = append(out, ip)
		}
	}
	return out
}

func listenKey(m model.Mapping) string {
	n := "tcp"
	if m.Proto == model.ProtoUDP {
		n = "udp"
	}
	return n + "|" + net.JoinHostPort(m.BindIP, strconv.Itoa(m.Port))
}

// apply reconciles listeners and per-mapping state with cfg.
func (b *Blacktip) apply(cfg *model.EdgeConfig) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.version = cfg.Version
	b.last = cfg
	b.authKey = cfg.AuthKey
	b.dashboard = cfg.Dashboard
	if b.dashboard == "" {
		b.dashboard = b.cfg.Hammerhead
	}
	if cfg.UpdateID != "" {
		b.requestUpdate(cfg.UpdateID)
	}
	b.local = map[string]bool{}
	for _, id := range cfg.Local {
		b.local[id] = true
	}
	if b.tr == nil {
		b.configureACME(cfg.AcmeEmail, cfg.AcmeStages)
	}

	b.configureTunnel(cfg)

	// per-mapping state: keep it (and its health history) when the definition is unchanged
	seen := map[string]bool{}
	for _, m := range cfg.Mappings {
		seen[m.ID] = true
		stable := m
		stable.AuthEpoch = 0
		sig, _ := json.Marshal(stable)
		if old := b.maps[m.ID]; old != nil {
			if old.sig == string(sig) {
				old.epoch.Store(m.AuthEpoch)
				continue
			}
			old.stop()
		}
		b.maps[m.ID] = b.newMapState(m, string(sig))
		b.maps[m.ID].epoch.Store(m.AuthEpoch)
	}
	for id, ms := range b.maps {
		if !seen[id] {
			ms.stop()
			delete(b.maps, id)
		}
	}

	want := map[string]*bindSet{}
	var webMaps []*mapState
	for _, m := range cfg.Mappings {
		if b.tr != nil && model.Web(m.Proto) {
			webMaps = append(webMaps, b.maps[m.ID]) // Traefik owns these ports
			continue
		}
		// a range opens one listener per port; each shifts the target port by its offset
		for _, ip := range bindIPs(m, cfg.Mappings) {
			for p := m.Port; p <= m.LastPort(); p++ {
				mm := m
				mm.BindIP, mm.Port = ip, p
				k := listenKey(mm)
				if want[k] == nil {
					want[k] = &bindSet{off: p - m.Port}
				}
				want[k].ms = append(want[k].ms, b.maps[m.ID])
			}
		}
	}
	// https ports can also answer on :80 with a redirect, unless something else owns that port
	redirects := map[string][]*mapState{}
	for _, m := range cfg.Mappings {
		if b.tr == nil && m.Proto == model.ProtoHTTPS && m.RedirectHTTP {
			for _, ip := range bindIPs(m, cfg.Mappings) {
				k := "tcp|" + net.JoinHostPort(ip, "80")
				if _, taken := want[k]; !taken {
					redirects[k] = append(redirects[k], b.maps[m.ID])
				}
			}
		}
	}

	for k, l := range b.listeners {
		_, w := want[k]
		_, rd := redirects[k]
		if !w && !rd {
			l.close()
			delete(b.listeners, k)
		}
	}
	b.addrSig = ""
	for _, m := range cfg.Mappings {
		if ips := bindIPs(m, cfg.Mappings); len(ips) != 1 || ips[0] != m.BindIP {
			b.addrSig = strings.Join(hostIPv4(), ",")
			break
		}
	}
	b.status = map[string]string{}
	bring := func(k string, ms []*mapState, off int, redirect bool) {
		if l, ok := b.listeners[k]; ok {
			if l.redirect == redirect && l.setMappings(ms, off) {
				return
			}
			l.close()
			delete(b.listeners, k)
		}
		_, addr, _ := strings.Cut(k, "|")
		host, _, _ := net.SplitHostPort(addr)
		l, err := b.open(host, ms, off, redirect)
		if err != nil {
			log.Printf("open %s: %v", k, err)
			if !redirect {
				port := k[strings.LastIndexByte(k, ':')+1:]
				for _, m := range ms {
					if prev := b.status[m.m.ID]; prev == "" {
						b.status[m.m.ID] = "port " + port + ": " + err.Error()
					} else if !strings.HasSuffix(prev, "(and others)") {
						b.status[m.m.ID] = prev + " (and others)"
					}
				}
			}
			return
		}
		b.listeners[k] = l
		if off == 0 {
			log.Printf("listening %s (%d mapping(s))", k, len(ms))
		}
	}
	for k, bs := range want {
		bring(k, bs.ms, bs.off, false)
	}
	for k, ms := range redirects {
		bring(k, ms, 0, true)
	}
	if b.tr != nil {
		binds := map[string][]string{}
		for _, ms := range webMaps {
			binds[ms.m.ID] = bindIPs(ms.m, cfg.Mappings)
		}
		b.tr.sync(b.ctx, webMaps, binds, cfg.AcmeEmail, cfg.AcmeStages)
		if e := b.tr.err(); e != "" {
			for _, ms := range webMaps {
				b.status[ms.m.ID] = e
			}
		}
	} else {
		b.warmCerts(cfg.Mappings)
	}
	go b.report()
}

// configureTunnel brings up (or updates) the WireGuard device. Caller holds b.mu.
func (b *Blacktip) configureTunnel(cfg *model.EdgeConfig) {
	if cfg.TunnelIP != b.tunIP && b.tun != nil {
		b.tun.Close()
		b.tun = nil
	}
	if b.tun == nil && cfg.TunnelIP != "" {
		addr, err := netip.ParseAddr(cfg.TunnelIP)
		if err != nil {
			log.Printf("bad tunnel address %q: %v", cfg.TunnelIP, err)
			return
		}
		dev, err := wg.New(b.wgKey, addr, tunnelPort(b.cfg.TunnelAddr))
		if err != nil {
			log.Printf("start WireGuard: %v", err)
			return
		}
		b.tun, b.tunIP = dev, cfg.TunnelIP
		log.Printf("WireGuard up at %s", cfg.TunnelIP)
		if ln, err := dev.ListenTCP(tunnel.MailTCPPort); err != nil {
			log.Printf("mail egress listener: %v", err)
		} else {
			go b.serveMail(ln)
		}
	}
	b.mail = map[string]model.MailPolicy{}
	for _, p := range cfg.MailEgress {
		b.mail[p.ID] = p
	}
	if sig := mailPolicySig(b.mail); sig != b.mailSig {
		b.mailSig = sig
		if len(b.mail) > 0 {
			go b.runMailChecks()
		}
	}
	b.peerIP = map[string]string{}
	var peers []wg.Peer
	for _, p := range cfg.Peers {
		b.peerIP[p.SiteID] = p.IP
		peers = append(peers, wg.Peer{PublicKey: p.PublicKey, AllowedIP: p.IP})
	}
	if b.tun != nil {
		if err := b.tun.SetPeers(peers); err != nil {
			log.Printf("peers: %v", err)
		}
	}
}

func (b *Blacktip) newMapState(m model.Mapping, sig string) *mapState {
	ctx, cancel := context.WithCancel(b.ctx)
	ms := &mapState{b: b, m: m, sig: sig, cancel: cancel}
	ms.health = make([]model.TargetHealth, len(m.Targets))
	for i := range ms.health {
		ms.health[i] = model.TargetHealth{Healthy: true, Detail: "not checked yet"}
	}
	if model.Web(m.Proto) && !m.IsRedirect() { // a redirect has nothing behind it to relay to or check
		if b.tr != nil {
			ms.startRelay()
		} else {
			ms.initProxy()
		}
	}
	if m.Proto != model.ProtoUDP && m.HealthSeconds != -1 && !m.IsRedirect() {
		go b.healthLoop(ctx, ms)
	}
	return ms
}

func (ms *mapState) stop() {
	ms.cancel()
	if ms.transport != nil {
		ms.transport.CloseIdleConnections()
	}
	if ms.relayLn != nil {
		ms.relayLn.Close()
	}
}

// retryLoop re-attempts listeners that failed to bind (port was busy, IP not up yet).
func (b *Blacktip) retryLoop(ctx context.Context) {
	t := time.NewTicker(20 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		b.mu.Lock()
		failed := false
		for _, e := range b.status {
			if e != "" {
				failed = true
			}
		}
		failed = failed || b.acmeFailing()
		last := b.last
		b.mu.Unlock()
		if failed && last != nil {
			b.apply(last)
		}
	}
}

func localIPs() []string {
	ips := []string{}
	addrs, _ := net.InterfaceAddrs()
	for _, a := range addrs {
		ipn, ok := a.(*net.IPNet)
		if !ok || ipn.IP.IsLoopback() || ipn.IP.IsLinkLocalUnicast() {
			continue
		}
		ips = append(ips, ipn.IP.String())
	}
	sort.Strings(ips)
	return ips
}

func (b *Blacktip) containers() []model.Container {
	if !b.cfg.Docker {
		return nil
	}
	b.dockerMu.Lock()
	defer b.dockerMu.Unlock()
	if time.Since(b.dockerAt) < 15*time.Second {
		return b.dockerList
	}
	ctx, cancel := context.WithTimeout(b.ctx, 5*time.Second)
	defer cancel()
	list, err := docker.List(ctx)
	b.dockerAt = time.Now()
	if err != nil {
		b.dockerList = nil
		return nil
	}
	b.dockerList = list
	return list
}

func (b *Blacktip) report() {
	b.reportMu.Lock()
	defer b.reportMu.Unlock()
	rep := model.EdgeReport{
		IPs: localIPs(), PublicKey: b.wgPub,
		Status: map[string]model.MappingStatus{}, Stats: map[string]model.Stat{},
		Containers: b.containers(), MailChecks: map[string]model.MailCheck{},
		Version: b.cfg.Version, AutoUpdate: b.cfg.AutoUpdate,
	}
	b.mu.Lock()
	rep.UpdateMsg, rep.UpdateID = b.updateMsg, b.updateID
	rep.UpdateStage, rep.UpdatePct, rep.UpdateLog = b.updStage, b.updPct, append([]string(nil), b.updLog...)
	for id, chk := range b.mailChecks {
		rep.MailChecks[id] = chk
	}
	for id, ms := range b.maps {
		st := model.MappingStatus{Error: b.status[id], Cert: b.certStatus(ms.m)}
		ms.mu.Lock()
		st.Targets = append([]model.TargetHealth(nil), ms.health...)
		ms.mu.Unlock()
		rep.Status[id] = st
		if c, in, out := ms.conns.Swap(0), ms.bytesIn.Swap(0), ms.bytesOut.Swap(0); c|in|out != 0 {
			rep.Stats[id] = model.Stat{Conns: c, BytesIn: in, BytesOut: out}
		}
	}
	b.mu.Unlock()
	body, _ := json.Marshal(rep)
	req, _ := http.NewRequest("POST", b.cfg.Hammerhead+"/edge/report", bytes.NewReader(body))
	req.Header.Set("Authorization", "Bearer "+b.cfg.Token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := b.http.Do(req)
	if err != nil {
		log.Printf("report: %v", err)
		// counters were swapped out; put them back so traffic isn't lost
		b.mu.Lock()
		for id, st := range rep.Stats {
			if ms := b.maps[id]; ms != nil {
				ms.conns.Add(st.Conns)
				ms.bytesIn.Add(st.BytesIn)
				ms.bytesOut.Add(st.BytesOut)
			}
		}
		b.mu.Unlock()
		return
	}
	resp.Body.Close()
}

func (b *Blacktip) reportLoop(ctx context.Context) {
	b.report()
	t := time.NewTicker(20 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			b.report()
			b.recheckAddresses()
		}
	}
}

// recheckAddresses re-applies the config when this machine gains or loses an address that a wildcard
// website is spread across, so the listeners follow the machine.
func (b *Blacktip) recheckAddresses() {
	b.mu.Lock()
	last, sig := b.last, b.addrSig
	b.mu.Unlock()
	if last == nil || sig == "" {
		return
	}
	if strings.Join(hostIPv4(), ",") != sig {
		b.apply(last)
	}
}

// ---------- access logs ----------

func (b *Blacktip) logAccess(e model.LogEntry) {
	select {
	case b.logq <- e:
	default: // drop rather than slow traffic down
	}
}

func (b *Blacktip) shipLogs(ctx context.Context) {
	t := time.NewTicker(3 * time.Second)
	defer t.Stop()
	var batch []model.LogEntry
	flush := func() {
		if len(batch) == 0 {
			return
		}
		body, _ := json.Marshal(batch)
		req, _ := http.NewRequest("POST", b.cfg.Hammerhead+"/edge/logs", bytes.NewReader(body))
		req.Header.Set("Authorization", "Bearer "+b.cfg.Token)
		req.Header.Set("Content-Type", "application/json")
		if resp, err := b.http.Do(req); err == nil {
			resp.Body.Close()
			batch = batch[:0]
		} else if len(batch) > 20000 {
			batch = batch[:0]
		}
	}
	for {
		select {
		case <-ctx.Done():
			return
		case e := <-b.logq:
			batch = append(batch, e)
			if len(batch) >= 200 {
				flush()
			}
		case <-t.C:
			flush()
		}
	}
}

// ---------- dialing ----------

// udpTunnel adapts a UDP socket inside the tunnel for one client session. Each
// datagram carries a one-byte type: 0 + target header (until the site has
// answered, so a lost first packet can't strand the session) or 1 for plain data.
type udpTunnel struct {
	net.Conn
	hdr  []byte
	seen atomic.Bool
}

func (u *udpTunnel) Write(p []byte) (int, error) {
	var out []byte
	if u.seen.Load() {
		out = append([]byte{1}, p...)
	} else {
		out = append([]byte{0, byte(len(u.hdr) >> 8), byte(len(u.hdr))}, u.hdr...)
		out = append(out, p...)
	}
	if _, err := u.Conn.Write(out); err != nil {
		return 0, err
	}
	return len(p), nil
}

func (u *udpTunnel) Read(p []byte) (int, error) {
	n, err := u.Conn.Read(p)
	if n > 0 {
		u.seen.Store(true)
	}
	return n, err
}

// dialTarget opens a connection to one target: through the WireGuard tunnel to
// the site's agent, or straight out of this machine when the site is the
// edge's built-in local one. tunneled is true for TCP streams that begin with
// a target header (the agent closes them if it can't reach the target).
func (b *Blacktip) dialTarget(t model.Target, proto string) (conn net.Conn, tunneled bool, err error) {
	b.mu.Lock()
	isLocal := b.local[t.SiteID]
	tun, ip := b.tun, b.peerIP[t.SiteID]
	b.mu.Unlock()
	if isLocal {
		d := net.Dialer{Timeout: 8 * time.Second}
		c, err := d.Dial(proto, net.JoinHostPort(t.Host, strconv.Itoa(t.Port)))
		return c, false, err
	}
	if tun == nil || ip == "" {
		return nil, false, errors.New("site is not connected")
	}
	hdr, _ := json.Marshal(tunnel.Header{Proto: proto, Host: t.Host, Port: t.Port})
	ctx, cancel := context.WithTimeout(b.ctx, 6*time.Second)
	defer cancel()
	if proto == "udp" {
		c, err := tun.DialContext(ctx, "udp", net.JoinHostPort(ip, strconv.Itoa(tunnel.ProxyUDPPort)))
		if err != nil {
			return nil, false, err
		}
		return &udpTunnel{Conn: c, hdr: hdr}, false, nil
	}
	c, err := tun.DialContext(ctx, "tcp", net.JoinHostPort(ip, strconv.Itoa(tunnel.ProxyTCPPort)))
	if err != nil {
		return nil, false, errors.New("site is not reachable through the tunnel")
	}
	if _, err := c.Write(append(hdr, '\n')); err != nil {
		c.Close()
		return nil, false, err
	}
	return c, true, nil
}

type bindSet struct {
	ms  []*mapState
	off int // added to every target port (non-zero only inside a port range)
}

// order lists target indexes: healthy ones first (rotated for round robin), then the rest.
func (ms *mapState) order() []int {
	n := len(ms.m.Targets)
	start := int((ms.rr.Add(1) - 1) % uint64(n))
	ms.mu.Lock()
	defer ms.mu.Unlock()
	good, bad := []int{}, []int{}
	for i := 0; i < n; i++ {
		idx := (start + i) % n
		if ms.health[idx].Healthy {
			good = append(good, idx)
		} else {
			bad = append(bad, idx)
		}
	}
	return append(good, bad...)
}

// dial connects to the best available target, failing over on errors.
// prefer >= 0 tries that target index first.
func (ms *mapState) dial(proto string, prefer int) (net.Conn, bool, error) {
	return ms.dialOff(proto, prefer, 0)
}

// dialOff is dial with every target port shifted by off (used for port ranges).
func (ms *mapState) dialOff(proto string, prefer, off int) (net.Conn, bool, error) {
	order := ms.order()
	if prefer >= 0 {
		rest := []int{prefer}
		for _, i := range order {
			if i != prefer {
				rest = append(rest, i)
			}
		}
		order = rest
	}
	var last error
	for _, i := range order {
		t := ms.m.Targets[i]
		t.Port += off
		c, tun, err := ms.b.dialTarget(t, proto)
		if err == nil {
			return c, tun, nil
		}
		last = err
	}
	return nil, false, last
}

// requestUpdate acts on "update now" from Hammerhead (id identifies the request). It only ever installs
// what this machine's own auto-update would (the newest release from its configured server), and only
// when auto-update is on. Caller holds b.mu.
func (b *Blacktip) requestUpdate(id string) {
	if id == b.updateID {
		return // already handled, or in progress
	}
	b.updateID, b.updateMsg = id, ""
	b.updStage, b.updPct, b.updLog = "starting", 0, nil
	b.updNote("Update requested from the dashboard")
	if !b.cfg.AutoUpdate || b.cfg.UpdateNow == nil {
		b.updateMsg = "auto-update is off on this machine (set BLACKTIP_AUTO_UPDATE=1 and restart it)"
		b.updStage = "failed"
		b.updNote("Refused: " + b.updateMsg)
		go b.report()
		return
	}
	if !b.updating.CompareAndSwap(false, true) {
		b.updateMsg = "another update is already running"
		b.updStage = "failed"
		b.updNote("Refused: " + b.updateMsg)
		go b.report()
		return
	}
	go func() {
		defer b.updating.Store(false)
		tag, err := b.cfg.UpdateNow(b.ctx, b.updProgress) // on success this restarts the process and does not return
		b.mu.Lock()
		switch {
		case err != nil:
			b.updateMsg = "update failed: " + err.Error()
		case tag == "":
			b.updateMsg = "already on the newest release"
		default:
			b.updateMsg = "updated to " + tag + ", restarting"
		}
		b.mu.Unlock()
		b.report()
	}()
}

// updNote adds a timestamped line to the update log. Caller holds b.mu.
func (b *Blacktip) updNote(line string) {
	b.updLog = append(b.updLog, time.Now().Format("15:04:05")+" "+line)
	if len(b.updLog) > 200 {
		b.updLog = b.updLog[len(b.updLog)-200:]
	}
}

// updProgress is called by the updater as it works; Hammerhead gets a fresh report at most once a
// second (and straight away for every new log line), so its progress bar follows along.
func (b *Blacktip) updProgress(stage string, pct int, line string) {
	b.mu.Lock()
	b.updStage = stage
	if pct >= 0 {
		b.updPct = pct
	}
	if line != "" {
		b.updNote(line)
	}
	send := line != "" || time.Since(b.updSent) > time.Second
	if send {
		b.updSent = time.Now()
	}
	b.mu.Unlock()
	if send {
		go b.report()
	}
}