Add the Blacktip edge
The edge: per-IP TCP/UDP/HTTP/HTTPS/TLS listeners with port ranges and PROXY protocol, WireGuard tunnel endpoint, logins in front of web ports (built-in proxy or Traefik), health checks and failover, and mail egress with DKIM signing and MTA-STS. Split out of the original single repository; depends on Reef for the shared types. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
VeillaxandClaude Sonnet 5.5 committed
1 parent
995ef9a6aa
commit
9448a532e9
17 files changed
+3492
-1
No files matched your search
+738
@@ -0,0 +1,738 @@
|
||||
// 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
|
||||
|
||||
// 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
|
||||
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)
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
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 p := m.Port; p <= m.LastPort(); p++ {
|
||||
mm := m
|
||||
mm.Port = 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 {
|
||||
k := "tcp|" + net.JoinHostPort(m.BindIP, "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.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)
|
||||
}
|
||||
l, err := b.open(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 {
|
||||
b.tr.sync(b.ctx, webMaps, 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) {
|
||||
if b.tr != nil {
|
||||
ms.startRelay()
|
||||
} else {
|
||||
ms.initProxy()
|
||||
}
|
||||
}
|
||||
if m.Proto != model.ProtoUDP && m.HealthSeconds != -1 {
|
||||
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{},
|
||||
}
|
||||
b.mu.Lock()
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 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
|
||||
}
|
||||
Reference in new issue
Block a user