From 9448a532e9e4075ff38cec185c54876ce527150f Mon Sep 17 00:00:00 2001 From: Veillax <126523099+Veillax@users.noreply.github.com> Date: Mon, 26 Oct 2026 02:45:16 -0600 Subject: [PATCH] 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 --- .env.example | 18 ++ .gitignore | 5 + README.md | 50 ++- blacktip.go | 738 +++++++++++++++++++++++++++++++++++++++++++ certs.go | 155 +++++++++ cmd/blacktip/main.go | 55 ++++ dkim.go | 113 +++++++ go.mod | 21 ++ go.sum | 24 ++ health.go | 132 ++++++++ listener.go | 421 ++++++++++++++++++++++++ mailegress.go | 510 ++++++++++++++++++++++++++++++ mtasts.go | 169 ++++++++++ mtasts_test.go | 107 +++++++ proxyproto.go | 65 ++++ traefik.go | 600 +++++++++++++++++++++++++++++++++++ web.go | 310 ++++++++++++++++++ 17 files changed, 3492 insertions(+), 1 deletion(-) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 blacktip.go create mode 100644 certs.go create mode 100644 cmd/blacktip/main.go create mode 100644 dkim.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 health.go create mode 100644 listener.go create mode 100644 mailegress.go create mode 100644 mtasts.go create mode 100644 mtasts_test.go create mode 100644 proxyproto.go create mode 100644 traefik.go create mode 100644 web.go diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..893cc1c --- /dev/null +++ b/.env.example @@ -0,0 +1,18 @@ +# Copy to .env (never commit it). Real environment variables win over this file. + +# Where Hammerhead is, and this edge's token (Server -> Edges -> Show command). +BLACKTIP_HAMMERHEAD=https://hammerhead.example.com +BLACKTIP_TOKEN= + +# UDP address for the WireGuard tunnel. +#BLACKTIP_TUNNEL=:51820 + +# Report local Docker containers so they can be picked as targets. +#BLACKTIP_DOCKER=1 + +# Serve http/https with Traefik (path to a Traefik v3 binary), and optionally +# use DNS challenge for wildcard certificates. The provider's credentials go here +# too, named exactly as Traefik documents for that provider. +#BLACKTIP_TRAEFIK=/usr/local/bin/traefik +#BLACKTIP_ACME_DNS=cloudflare +#CF_DNS_API_TOKEN= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7bf8720 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +bin/ +*-data/ +.env +wg.key +*.db diff --git a/README.md b/README.md index 2106d04..e4436e9 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,51 @@ # Blacktip -Blacktip runs on your public server. It owns the IP addresses and opens the TCP, UDP, HTTP and HTTPS ports you pick, including the same port on different IPs. It terminates the WireGuard tunnels, puts logins in front of websites (optionally through Traefik), checks target health, and delivers your outgoing mail with DKIM signing and MTA-STS. \ No newline at end of file +Blacktip runs on your public server. It owns the IP addresses and opens the TCP, UDP, HTTP and HTTPS ports you pick, including the same port on different IPs. It terminates the WireGuard tunnels, puts logins in front of websites (optionally through Traefik), checks target health, and delivers your outgoing mail with DKIM signing and MTA-STS. + +Part of the sharks stack: [Hammerhead](https://git.veillax.com/Veillax/hammerhead) (control plane and dashboard) · **Blacktip** (edge) · [Bamboo](https://git.veillax.com/Veillax/bamboo) (site agent) · [Reef](https://git.veillax.com/Veillax/reef) (shared library). Everything is configured from Hammerhead's web dashboard; the full documentation lives in [its README](https://git.veillax.com/Veillax/hammerhead). + +## Run it + +In Hammerhead, go to Server → Edges → add an edge. The dashboard shows the exact command with your token. It looks like this: + +``` +blacktip -hammerhead https://hammerhead.example.com -token -tunnel :51820 -docker +``` + +Open the WireGuard UDP port (default 51820) and any ports you open from the dashboard in this machine's firewall. Blacktip only needs to reach Hammerhead over HTTPS; it never needs inbound access from it. + +| Flag | Environment | Meaning | +|---|---|---| +| `-hammerhead` | `BLACKTIP_HAMMERHEAD` | URL of Hammerhead | +| `-token` | `BLACKTIP_TOKEN` | the edge's token | +| `-tunnel` | `BLACKTIP_TUNNEL` | UDP listen address for WireGuard (default `:51820`) | +| `-data` | `BLACKTIP_DATA` | folder for the WireGuard key and certificates | +| `-docker` | `BLACKTIP_DOCKER=1` | report local Docker containers so they can be picked as targets | +| `-traefik` | `BLACKTIP_TRAEFIK` | path to a Traefik v3 binary, to serve http/https with Traefik | +| `-acme-dns` | `BLACKTIP_ACME_DNS` | with Traefik: DNS-challenge provider for wildcard certificates | +| `-no-acme` | | with Traefik: serve its self-signed certificate instead of Let's Encrypt | +| `-env` | | `.env` file to read (default `./.env`) | + +Settings and secrets can live in a `.env` file instead of the command line; see `.env.example`. Real environment variables always win. + +## What it does + +- **Ports:** TCP, UDP, HTTP, HTTPS and TLS passthrough, one listener per IP and port, opened and closed live as you change them in the dashboard. Port ranges for TCP and UDP. Round-robin load balancing and failover across several targets, with health checks. +- **PROXY protocol (v1/v2)** on TCP and TLS ports, so the target sees the real client address. +- **Web:** automatic Let's Encrypt certificates, HTTP/2, websockets, and logins (account, password, PIN, emailed code, share links, path and IP rules) enforced in front of any site. Use the built-in proxy, or `-traefik` to let Traefik serve http/https (raw TCP, UDP and TLS-passthrough ports always stay with Blacktip, since Traefik can't open ports without a restart). +- **Mail egress:** delivers your mail server's outgoing mail straight to the recipient's server from a public IP you choose, with the HELO name you set, opportunistic STARTTLS, optional DKIM signing, MTA-STS enforcement, a per-minute send limit and a deliverability checklist. +- **Tunnel:** WireGuard through a userspace network stack, so no root, no kernel interface and no driver. It works on Windows, macOS and Linux, and sites behind NAT connect out to it. + +## Build and test + +``` +go env -w GOPRIVATE=git.veillax.com # once, so Go fetches straight from this server +go build -o bin/ ./cmd/blacktip +go test ./... +``` + +The end-to-end tests that run Blacktip together with Hammerhead and Bamboo live in the Hammerhead repository. + +## License + +AGPL-3.0. See `LICENSE`. diff --git a/blacktip.go b/blacktip.go new file mode 100644 index 0000000..7cb136e --- /dev/null +++ b/blacktip.go @@ -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 +} diff --git a/certs.go b/certs.go new file mode 100644 index 0000000..da77605 --- /dev/null +++ b/certs.go @@ -0,0 +1,155 @@ +package blacktip + +import ( + "context" + "crypto/tls" + "errors" + "path/filepath" + "strings" + "time" + + "git.veillax.com/Veillax/reef/model" + "golang.org/x/crypto/acme" + "golang.org/x/crypto/acme/autocert" +) + +// Certificates for https ports come from Let's Encrypt via autocert, using the +// TLS-ALPN-01 challenge on the same port 443 listener, so nothing extra has to +// be opened. A wildcard hostname such as *.example.com gets a separate +// certificate for each subdomain on its first visit. + +type acmeState struct { + mgr *autocert.Manager + email string + staging bool + tried map[string]time.Time + certs map[string]string +} + +// configureACME builds the manager. Caller holds b.mu. +func (b *Blacktip) configureACME(email string, staging bool) { + a := &b.acme + if a.tried == nil { + a.tried, a.certs = map[string]time.Time{}, map[string]string{} + } + if a.mgr != nil && a.email == email && a.staging == staging { + return + } + m := &autocert.Manager{ + Prompt: autocert.AcceptTOS, + Cache: autocert.DirCache(filepath.Join(b.cfg.DataDir, "certs")), + HostPolicy: b.hostPolicy, + Email: email, + } + if staging { + m.Client = &acme.Client{DirectoryURL: "https://acme-staging-v02.api.letsencrypt.org/directory"} + } + a.mgr, a.email, a.staging = m, email, staging + a.tried = map[string]time.Time{} +} + +// hostPolicy only lets us request certificates for hostnames we actually serve over https. +func (b *Blacktip) hostPolicy(_ context.Context, host string) error { + b.mu.Lock() + defer b.mu.Unlock() + for _, ms := range b.maps { + if ms.m.Proto == model.ProtoHTTPS && model.HostMatches(ms.m.Hostname, strings.ToLower(host)) { + return nil + } + } + return errors.New("not a hostname served here") +} + +func (b *Blacktip) getCert(hello *tls.ClientHelloInfo) (*tls.Certificate, error) { + if b.cfg.GetCertificate != nil { + return b.cfg.GetCertificate(hello) + } + b.mu.Lock() + mgr := b.acme.mgr + b.mu.Unlock() + if mgr == nil { + return nil, errors.New("certificates are not configured yet") + } + return mgr.GetCertificate(hello) +} + +// warmCerts requests certificates ahead of the first visitor so problems show up in the UI. +// Caller holds b.mu. +func (b *Blacktip) warmCerts(ms []model.Mapping) { + for _, m := range ms { + if m.Proto != model.ProtoHTTPS || strings.HasPrefix(m.Hostname, "*") { + continue + } + host := m.Hostname + if b.cfg.GetCertificate != nil { + b.acme.certs[host] = "custom certificate" + continue + } + wait := 12 * time.Hour + if strings.HasPrefix(b.acme.certs[host], "error") { + wait = 3 * time.Minute + } + if t, ok := b.acme.tried[host]; ok && time.Since(t) < wait { + continue + } + b.acme.tried[host] = time.Now() + b.acme.certs[host] = "requesting…" + mgr := b.acme.mgr + go func() { + ctx, cancel := context.WithTimeout(b.ctx, 2*time.Minute) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := mgr.GetCertificate(&tls.ClientHelloInfo{ServerName: host}) + done <- err + }() + var err error + select { + case err = <-done: + case <-ctx.Done(): + err = ctx.Err() + } + b.mu.Lock() + if err != nil { + b.acme.certs[host] = "error: " + err.Error() + } else { + b.acme.certs[host] = "valid" + } + b.mu.Unlock() + go b.report() + }() + } +} + +// certStatus describes the certificate for m. Caller holds b.mu. +func (b *Blacktip) certStatus(m model.Mapping) string { + if m.Proto != model.ProtoHTTPS { + return "" + } + if strings.HasPrefix(m.Hostname, "*") { + if b.tr != nil { + if b.cfg.Traefik.NoACME { + return "self-signed (Traefik default certificate)" + } + if b.cfg.Traefik.ACMEDNS == "" { + return "error: wildcard certificates need -acme-dns " + } + return "wildcard certificate via DNS challenge" + } + return "issued per subdomain on first visit" + } + if s := b.acme.certs[m.Hostname]; s != "" { + return s + } + return "pending" +} + +// acmeFailing reports whether any certificate request failed. Caller holds b.mu. +func (b *Blacktip) acmeFailing() bool { + for _, s := range b.acme.certs { + if strings.HasPrefix(s, "error") { + return true + } + } + return false +} diff --git a/cmd/blacktip/main.go b/cmd/blacktip/main.go new file mode 100644 index 0000000..d017d81 --- /dev/null +++ b/cmd/blacktip/main.go @@ -0,0 +1,55 @@ +// Blacktip is the edge: run it on a public server that owns the IP addresses. +package main + +import ( + "context" + "flag" + "log" + "os" + "os/signal" + "strings" + + "git.veillax.com/Veillax/blacktip" + "git.veillax.com/Veillax/reef/dotenv" +) + +func env(k, def string) string { + if v := os.Getenv(k); v != "" { + return v + } + return def +} + +func main() { + // secrets (tokens, DNS-provider credentials) can live in a .env file instead of the command line + envFile := dotenv.ArgPath(os.Args[1:], ".env") + if _, err := dotenv.Load(envFile); err != nil { + log.Fatalf("read %s: %v", envFile, err) + } + hh := flag.String("hammerhead", env("BLACKTIP_HAMMERHEAD", ""), "URL of hammerhead, e.g. http://10.0.0.5:8080") + token := flag.String("token", env("BLACKTIP_TOKEN", ""), "edge token from the hammerhead UI") + tun := flag.String("tunnel", env("BLACKTIP_TUNNEL", ":51820"), "UDP listen address for the WireGuard tunnel (open this UDP port in your firewall)") + data := flag.String("data", env("BLACKTIP_DATA", "./blacktip-data"), "directory for the WireGuard key and certificates") + dockerOn := flag.Bool("docker", os.Getenv("BLACKTIP_DOCKER") == "1", "report local Docker containers so they can be picked as targets in the UI") + traefikPath := flag.String("traefik", os.Getenv("BLACKTIP_TRAEFIK"), "path to a Traefik binary: serve http/https resources with Traefik instead of the built-in proxy") + acmeDNS := flag.String("acme-dns", os.Getenv("BLACKTIP_ACME_DNS"), "with -traefik: DNS-challenge provider (e.g. cloudflare) for wildcard certificates; its credentials go in this process's environment") + noACME := flag.Bool("no-acme", false, "with -traefik: serve Traefik's self-signed certificate instead of requesting Let's Encrypt ones") + flag.String("env", ".env", "file to read secrets and settings from") + flag.Parse() + if *hh == "" || *token == "" { + flag.Usage() + os.Exit(2) + } + b, err := blacktip.New(blacktip.Config{ + Hammerhead: strings.TrimRight(*hh, "/"), Token: *token, TunnelAddr: *tun, DataDir: *data, Docker: *dockerOn, + Traefik: blacktip.TraefikConfig{Path: *traefikPath, ACMEDNS: *acmeDNS, NoACME: *noACME}, + }) + if err != nil { + log.Fatal(err) + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + if err := b.Run(ctx); err != nil { + log.Fatal(err) + } +} diff --git a/dkim.go b/dkim.go new file mode 100644 index 0000000..f5e4683 --- /dev/null +++ b/dkim.go @@ -0,0 +1,113 @@ +package blacktip + +import ( + "bytes" + "crypto" + "crypto/x509" + "encoding/pem" + "net" + "net/mail" + "regexp" + "strings" + + "git.veillax.com/Veillax/reef/model" + "github.com/emersion/go-msgauth/dkim" +) + +func netLookupTXT(name string) ([]string, error) { return net.LookupTXT(name) } + +var dkimHeaders = []string{"From", "To", "Cc", "Subject", "Date", "Message-ID", "MIME-Version", "Content-Type", + "Content-Transfer-Encoding", "Reply-To", "In-Reply-To", "References"} + +var dkimDomainRE = regexp.MustCompile(`(?i)(?:^|[;\s])d=\s*([^;\s]+)`) + +// toCRLF makes every line end in CRLF. Mail travels as CRLF and DKIM is defined +// over that form; the relay hands us the message with plain LF. +func toCRLF(b []byte) []byte { + b = bytes.ReplaceAll(b, []byte("\r\n"), []byte("\n")) + return bytes.ReplaceAll(b, []byte("\n"), []byte("\r\n")) +} + +// pickSigner finds the key for a sender domain: an exact match, else the +// longest parent domain (mail from news.example.com may be signed by example.com). +func pickSigner(signers []model.DKIMSigner, domain string) *model.DKIMSigner { + var best *model.DKIMSigner + for i := range signers { + d := strings.ToLower(signers[i].Domain) + if domain == d || strings.HasSuffix(domain, "."+d) { + if best == nil || len(d) > len(best.Domain) { + best = &signers[i] + } + } + } + return best +} + +// sign adds a DKIM-Signature for the message's From domain if the policy has a +// key for it. It returns the bytes to send, and a short note for the log. +// Any problem results in the message being sent unsigned, never dropped. +func (s *mailSession) sign(data []byte) ([]byte, string) { + if len(s.pol.DKIM) == 0 { + return data, "" + } + msg := toCRLF(data) + m, err := mail.ReadMessage(bytes.NewReader(msg)) + if err != nil { + return data, "unsigned: unreadable headers" + } + domain := "" + if a, err := mail.ParseAddress(m.Header.Get("From")); err == nil { + domain = domainOf(a.Address) + } + if domain == "" { + domain = domainOf(s.from) + } + sg := pickSigner(s.pol.DKIM, domain) + if sg == nil { + return data, "" + } + for _, existing := range m.Header["Dkim-Signature"] { // already signed for this domain by your mail server: leave it + if mm := dkimDomainRE.FindStringSubmatch(existing); mm != nil && strings.EqualFold(mm[1], sg.Domain) { + return data, "already signed" + } + } + block, _ := pem.Decode([]byte(sg.PrivateKey)) + if block == nil { + return data, "unsigned: bad key" + } + key, err := x509.ParsePKCS8PrivateKey(block.Bytes) + signer, ok := key.(crypto.Signer) + if err != nil || !ok { + return data, "unsigned: bad key" + } + var out bytes.Buffer + err = dkim.Sign(&out, bytes.NewReader(msg), &dkim.SignOptions{ + Domain: sg.Domain, Selector: sg.Selector, Signer: signer, Hash: crypto.SHA256, + HeaderCanonicalization: dkim.CanonicalizationRelaxed, BodyCanonicalization: dkim.CanonicalizationRelaxed, + HeaderKeys: dkimHeaders, + }) + if err != nil { + return data, "unsigned: " + err.Error() + } + return out.Bytes(), "dkim " + sg.Selector + "._domainkey." + sg.Domain +} + +// checkDKIM looks each key up in DNS the way a receiving server would. +func (b *Blacktip) checkDKIM(signers []model.DKIMSigner) map[string]string { + out := map[string]string{} + for _, sg := range signers { + txts, err := b.lookupTXT(sg.Selector + "._domainkey." + sg.Domain) + want := "p=" + sg.PublicKey + status := "missing" + if err == nil && len(txts) > 0 { + status = "mismatch" + for _, t := range txts { + if strings.Contains(strings.ReplaceAll(t, " ", ""), want) { + status = "published" + } + } + } + out[sg.KeyID] = status + } + return out +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..6c9ddcb --- /dev/null +++ b/go.mod @@ -0,0 +1,21 @@ +module git.veillax.com/Veillax/blacktip + +go 1.27.3 + +require ( + git.veillax.com/Veillax/reef v0.1.0 + github.com/emersion/go-msgauth v0.7.0 + golang.org/x/crypto v0.58.0 +) + +require ( + github.com/Microsoft/go-winio v0.6.4 // indirect + github.com/google/btree v1.1.2 // indirect + golang.org/x/net v0.61.0 // indirect + golang.org/x/sys v0.49.0 // indirect + golang.org/x/text v0.43.0 // indirect + golang.org/x/time v0.7.0 // indirect + golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect + golang.zx2c4.com/wireguard v0.0.0-20261024220952-b3bc704ef0ee // indirect + gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..d78a590 --- /dev/null +++ b/go.sum @@ -0,0 +1,24 @@ +git.veillax.com/Veillax/reef v0.1.0 h1:O0OhEKFx2CsvACJ20sfzJNjGsngomtfDdKdxu4ROpCs= +git.veillax.com/Veillax/reef v0.1.0/go.mod h1:ZBUd/BARO5kkpUNTP81TTY1aQXLHFR6bKDhCxckjHso= +github.com/Microsoft/go-winio v0.6.4 h1:kJF2pamtYCWXf0xAT8sAkwKC0pMl2HYBrRPHAqyUc+g= +github.com/Microsoft/go-winio v0.6.4/go.mod h1:1xl5JXG4ze/m74GH6Dc4Mwxfc3VL5siRVQjIAlHH07U= +github.com/emersion/go-msgauth v0.7.0 h1:vj2hMn6KhFtW41kshIBTXvp6KgYSqpA/ZN9Pv4g1INc= +github.com/emersion/go-msgauth v0.7.0/go.mod h1:mmS9I6HkSovrNgq0HNXTeu8l3sRAAuQ9RMvbM4KU7Ck= +github.com/google/btree v1.1.2 h1:xf4v41cLI2Z6FxbKm+8Bu+m8ifhj15JuZ9sa0jZCMUU= +github.com/google/btree v1.1.2/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= +golang.org/x/crypto v0.58.0 h1:COkYLr4k7nDI4r6QtMML2AfzeIswWHAJthLS6K/M/54= +golang.org/x/crypto v0.58.0/go.mod h1:Xh+kl5A+M0gMiWDytH2Hhr5Z7WbNCpv/eB/EFgfSk+w= +golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= +golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= +golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= +golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ= +golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 h1:B82qJJgjvYKsXS9jeunTOisW56dUokqW/FOteYJJ/yg= +golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2/go.mod h1:deeaetjYA+DHMHg+sMSMI58GrEteJUUzzw7en6TJQcI= +golang.zx2c4.com/wireguard v0.0.0-20261024220952-b3bc704ef0ee h1:hByBPN/P/BVzJ591rJfvNhdoL94vmBtRbnVw77QJL20= +golang.zx2c4.com/wireguard v0.0.0-20261024220952-b3bc704ef0ee/go.mod h1:rpwXGsirqLqN2L0JDJQlwOboGHmptD5ZD6T2VmcqhTw= +gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c h1:m/r7OM+Y2Ty1sgBQ7Qb27VgIMBW8ZZhT4gLnUyDIhzI= +gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c/go.mod h1:3r5CMtNQMKIvBlrmM9xWUNamjKBYPOWyXOjmg5Kts3g= diff --git a/health.go b/health.go new file mode 100644 index 0000000..ac934f0 --- /dev/null +++ b/health.go @@ -0,0 +1,132 @@ +package blacktip + +import ( + "context" + "crypto/tls" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "time" + + "git.veillax.com/Veillax/reef/model" +) + +// healthLoop checks every target of ms until ctx is cancelled. Routing prefers +// healthy targets, so a dead backend is skipped without anyone editing config. +func (b *Blacktip) healthLoop(ctx context.Context, ms *mapState) { + every := time.Duration(ms.m.HealthSeconds) * time.Second + if ms.m.HealthSeconds <= 0 { + every = 30 * time.Second + } + run := func() { + results := make([]model.TargetHealth, len(ms.m.Targets)) + var wg sync.WaitGroup + for i := range ms.m.Targets { + wg.Add(1) + go func(i int) { + defer wg.Done() + results[i] = ms.check(ctx, i) + }(i) + } + wg.Wait() + if ctx.Err() != nil { + return + } + ms.mu.Lock() + changed := false + for i, r := range results { + if ms.health[i].Healthy != r.Healthy { + changed = true + } + } + ms.health = results + ms.mu.Unlock() + if changed { + go b.report() + } + } + // healthy targets are re-checked every `every`; unhealthy ones every few + // seconds so recovery (or a site that was still connecting) shows up fast + run() + for { + wait := every + ms.mu.Lock() + for _, h := range ms.health { + if !h.Healthy && wait > 5*time.Second { + wait = 5 * time.Second + } + } + ms.mu.Unlock() + select { + case <-ctx.Done(): + return + case <-time.After(wait): + run() + } + } +} + +func (ms *mapState) check(ctx context.Context, i int) model.TargetHealth { + t := ms.m.Targets[i] + if model.Web(ms.m.Proto) && ms.m.HealthPath != "" { + return ms.checkHTTP(ctx, t) + } + c, tunneled, err := ms.b.dialTarget(t, "tcp") + if err != nil { + return model.TargetHealth{Detail: err.Error()} + } + defer c.Close() + if tunneled { + // A tunnel stream opens even when the far side can't connect, but the + // agent then closes it straight away. Silence means the target is up. + c.SetReadDeadline(time.Now().Add(800 * time.Millisecond)) + one := make([]byte, 1) + if _, err := c.Read(one); err != nil { + if ne, ok := err.(net.Error); ok && ne.Timeout() { + return model.TargetHealth{Healthy: true} + } + if err == io.EOF || strings.Contains(err.Error(), "closed") || strings.Contains(err.Error(), "reset") { + return model.TargetHealth{Detail: "the site couldn't connect to the target"} + } + return model.TargetHealth{Detail: err.Error()} + } + } + return model.TargetHealth{Healthy: true} +} + +func (ms *mapState) checkHTTP(ctx context.Context, t model.Target) model.TargetHealth { + scheme := "http" + if ms.m.BackendHTTPS { + scheme = "https" + } + tr := &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + c, _, err := ms.b.dialTarget(t, "tcp") + return c, err + }, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DisableKeepAlives: true, + } + defer tr.CloseIdleConnections() + ctx, cancel := context.WithTimeout(ctx, 6*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, "GET", scheme+"://target"+ms.m.HealthPath, nil) + if ms.m.RewriteHost != "" { + req.Host = ms.m.RewriteHost + } else if h := strings.TrimPrefix(ms.m.Hostname, "*."); h != "" { + req.Host = h + } + req.Header.Set("User-Agent", "blacktip-healthcheck") + resp, err := (&http.Client{Transport: tr, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}).Do(req) + if err != nil { + return model.TargetHealth{Detail: err.Error()} + } + resp.Body.Close() + if resp.StatusCode >= 200 && resp.StatusCode < 400 { + return model.TargetHealth{Healthy: true} + } + return model.TargetHealth{Detail: fmt.Sprintf("health check got HTTP %d", resp.StatusCode)} +} diff --git a/listener.go b/listener.go new file mode 100644 index 0000000..5b9042f --- /dev/null +++ b/listener.go @@ -0,0 +1,421 @@ +package blacktip + +import ( + "crypto/tls" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "git.veillax.com/Veillax/reef/model" + "git.veillax.com/Veillax/reef/tunnel" + "golang.org/x/crypto/acme" +) + +// listener is one bound ip:port. For tcp-class kinds it can serve several +// mappings (hostname routing); udp and plain tcp have exactly one. +type listener struct { + b *Blacktip + kind string // tcp, udp, http, https, tls + redirect bool // an http->https redirect helper rather than a real mapping + off int // port range: this listener's offset from the range start, added to target ports + tcp net.Listener + udp net.PacketConn + srv *http.Server + + mu sync.Mutex + maps []*mapState + done chan struct{} + sessions map[string]*udpSession +} + +func (b *Blacktip) open(ms []*mapState, off int, redirect bool) (*listener, error) { + m := ms[0].m + addr := net.JoinHostPort(m.BindIP, strconv.Itoa(m.Port+off)) + l := &listener{b: b, kind: m.Proto, redirect: redirect, maps: ms, off: off, done: make(chan struct{}), sessions: map[string]*udpSession{}} + if redirect { + addr = net.JoinHostPort(m.BindIP, "80") + l.kind = model.ProtoHTTP + } + if m.Proto == model.ProtoUDP { + pc, err := net.ListenPacket("udp", addr) + if err != nil { + return nil, err + } + l.udp = pc + go l.serveUDP() + go l.reapUDP() + return l, nil + } + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, err + } + l.tcp = ln + switch { + case redirect: + l.srv = &http.Server{Handler: http.HandlerFunc(l.serveRedirect), ReadHeaderTimeout: 15 * time.Second} + go l.srv.Serve(ln) + case model.Web(m.Proto): + l.srv = &http.Server{Handler: l, ReadHeaderTimeout: 15 * time.Second, IdleTimeout: 120 * time.Second, + ErrorLog: log.New(io.Discard, "", 0)} + serve := ln + if m.Proto == model.ProtoHTTPS { + serve = tls.NewListener(ln, &tls.Config{ + GetCertificate: b.getCert, + NextProtos: []string{"h2", "http/1.1", acme.ALPNProto}, + MinVersion: tls.VersionTLS12, + }) + } + go l.srv.Serve(serve) + default: + go l.serveTCP() + } + return l, nil +} + +// setMappings swaps the routing table; it returns false if the listener must be restarted. +func (l *listener) setMappings(ms []*mapState, off int) bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.off != off { + return false + } + if l.kind == model.ProtoUDP && !l.redirect { + if l.maps[0].m.Targets[0] != ms[0].m.Targets[0] { + return false + } + } + l.maps = ms + return true +} + +func (l *listener) snapshot() []*mapState { + l.mu.Lock() + defer l.mu.Unlock() + return l.maps +} + +func (l *listener) close() { + select { + case <-l.done: + return + default: + close(l.done) + } + if l.srv != nil { + l.srv.Close() + } + if l.tcp != nil { + l.tcp.Close() + } + if l.udp != nil { + l.udp.Close() + } + l.mu.Lock() + for _, s := range l.sessions { + s.conn.Close() + } + l.mu.Unlock() +} + +// route picks the mapping for host: exact match first, then wildcards. +func route(ms []*mapState, host string) (*mapState, bool) { + host = strings.ToLower(strings.TrimSuffix(host, ".")) + for _, m := range ms { + if m.m.Hostname == host { + return m, true + } + } + for _, m := range ms { + if model.HostMatches(m.m.Hostname, host) { + return m, true + } + } + return nil, false +} + +func hostOnly(h string) string { + if hh, _, err := net.SplitHostPort(h); err == nil { + h = hh + } + return strings.ToLower(strings.Trim(h, "[]")) +} + +// serveRedirect answers on port 80 for https ports that asked for it. +func (l *listener) serveRedirect(w http.ResponseWriter, r *http.Request) { + host := hostOnly(r.Host) + ms, ok := route(l.snapshot(), host) + if !ok { + http.NotFound(w, r) + return + } + target := "https://" + host + if ms.m.Port != 443 { + target += fmt.Sprintf(":%d", ms.m.Port) + } + http.Redirect(w, r, target+r.URL.RequestURI(), http.StatusPermanentRedirect) +} + +// ---------- raw TCP and TLS passthrough ---------- + +func (l *listener) serveTCP() { + for { + c, err := l.tcp.Accept() + if err != nil { + select { + case <-l.done: + return + default: + time.Sleep(50 * time.Millisecond) + continue + } + } + go l.handleTCP(c) + } +} + +func (l *listener) handleTCP(c net.Conn) { + start := time.Now() + maps := l.snapshot() + ms := maps[0] + var first []byte + host := "" + if l.kind == model.ProtoTLS { + c.SetReadDeadline(time.Now().Add(10 * time.Second)) + var err error + first, host, err = peekSNI(c) + c.SetReadDeadline(time.Time{}) + if err != nil { + c.Close() + return + } + var ok bool + if ms, ok = route(maps, host); !ok { + c.Close() + return + } + } + entry := model.LogEntry{Time: start, Mapping: ms.m.ID, Remote: c.RemoteAddr().String(), Host: host, Result: "connection"} + up, _, err := ms.dialOff("tcp", -1, l.off) + if err != nil { + entry.Result = "no-target: " + err.Error() + l.b.logAccess(entry) + c.Close() + return + } + if v := ms.m.ProxyProtocol; v != "" { + // the real client address goes first, before any byte the client sent + if _, err := up.Write(proxyHeader(v, c.RemoteAddr(), c.LocalAddr())); err != nil { + up.Close() + c.Close() + return + } + } + if len(first) > 0 { + if _, err := up.Write(first); err != nil { + up.Close() + c.Close() + return + } + } + in, out := tunnel.Proxy(c, c, up, up) + in += int64(len(first)) + ms.conns.Add(1) + ms.bytesIn.Add(in) + ms.bytesOut.Add(out) + entry.BytesIn, entry.BytesOut, entry.Millis = in, out, time.Since(start).Milliseconds() + l.b.logAccess(entry) +} + +// peekSNI reads one TLS record holding the ClientHello and extracts the server name. +func peekSNI(c net.Conn) ([]byte, string, error) { + hdr := make([]byte, 5) + if _, err := io.ReadFull(c, hdr); err != nil { + return nil, "", err + } + if hdr[0] != 0x16 { + return hdr, "", errors.New("not TLS") + } + n := int(hdr[3])<<8 | int(hdr[4]) + if n > 1<<14 { + return hdr, "", errors.New("record too large") + } + body := make([]byte, n) + if _, err := io.ReadFull(c, body); err != nil { + return append(hdr, body...), "", err + } + all := append(hdr, body...) + name, err := parseSNI(body) + return all, name, err +} + +func parseSNI(b []byte) (string, error) { + bad := errors.New("malformed ClientHello") + skip := func(n int) bool { + if len(b) < n { + return false + } + b = b[n:] + return true + } + u8 := func() (int, bool) { + if len(b) < 1 { + return 0, false + } + v := int(b[0]) + b = b[1:] + return v, true + } + u16 := func() (int, bool) { + if len(b) < 2 { + return 0, false + } + v := int(b[0])<<8 | int(b[1]) + b = b[2:] + return v, true + } + if len(b) < 4 || b[0] != 0x01 { + return "", bad + } + if !skip(4 + 2 + 32) { // handshake header, version, random + return "", bad + } + n, ok := u8() + if !ok || !skip(n) { // session id + return "", bad + } + n, ok = u16() + if !ok || !skip(n) { // cipher suites + return "", bad + } + n, ok = u8() + if !ok || !skip(n) { // compression + return "", bad + } + n, ok = u16() + if !ok || n > len(b) { + return "", bad + } + b = b[:n] + for len(b) >= 4 { + typ, _ := u16() + l, _ := u16() + if l > len(b) { + return "", bad + } + ext := b[:l] + b = b[l:] + if typ != 0 { // server_name + continue + } + if len(ext) < 5 { + return "", bad + } + nl := int(ext[3])<<8 | int(ext[4]) + if 5+nl > len(ext) { + return "", bad + } + return strings.ToLower(string(ext[5 : 5+nl])), nil + } + return "", errors.New("no SNI in ClientHello") +} + +// ---------- UDP ---------- + +type udpSession struct { + conn net.Conn + last time.Time +} + +const udpIdle = 60 * time.Second + +func (l *listener) serveUDP() { + buf := make([]byte, 65535) + for { + n, addr, err := l.udp.ReadFrom(buf) + if err != nil { + select { + case <-l.done: + return + default: + time.Sleep(50 * time.Millisecond) + continue + } + } + key := addr.String() + ms := l.snapshot()[0] + l.mu.Lock() + s := l.sessions[key] + l.mu.Unlock() + if s == nil { + conn, _, err := ms.dialOff("udp", -1, l.off) + if err != nil { + continue + } + s = &udpSession{conn: conn} + l.mu.Lock() + l.sessions[key] = s + l.mu.Unlock() + ms.conns.Add(1) + go l.pumpBack(ms, key, addr, s) + } + l.mu.Lock() + s.last = time.Now() + l.mu.Unlock() + ms.bytesIn.Add(int64(n)) + s.conn.Write(buf[:n]) + } +} + +// pumpBack relays replies from the target to the UDP client until the session ends. +func (l *listener) pumpBack(ms *mapState, key string, addr net.Addr, s *udpSession) { + defer func() { + s.conn.Close() + l.mu.Lock() + if l.sessions[key] == s { + delete(l.sessions, key) + } + l.mu.Unlock() + }() + buf := make([]byte, 65535) + for { + n, err := s.conn.Read(buf) + if err != nil { + return + } + p := buf[:n] + if _, err := l.udp.WriteTo(p, addr); err != nil { + return + } + ms.bytesOut.Add(int64(len(p))) + l.mu.Lock() + s.last = time.Now() + l.mu.Unlock() + } +} + +func (l *listener) reapUDP() { + t := time.NewTicker(15 * time.Second) + defer t.Stop() + for { + select { + case <-l.done: + return + case <-t.C: + } + l.mu.Lock() + for _, s := range l.sessions { + if time.Since(s.last) > udpIdle { + s.conn.Close() // pumpBack removes it + } + } + l.mu.Unlock() + } +} diff --git a/mailegress.go b/mailegress.go new file mode 100644 index 0000000..3892024 --- /dev/null +++ b/mailegress.go @@ -0,0 +1,510 @@ +package blacktip + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/smtp" + "net/textproto" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.veillax.com/Veillax/reef/model" + "git.veillax.com/Veillax/reef/netguard" + "git.veillax.com/Veillax/reef/tunnel" +) + +// Mail egress: a site's SMTP relay hands this edge one mail transaction at a +// time through the WireGuard tunnel. The edge finds the recipient's mail server +// (MX lookup, or the configured smarthost), connects from the policy's source +// IP, introduces itself with the policy's HELO name, and relays the exchange, +// passing the receiving server's replies straight back so the sender's own +// queue and retry logic keep working. +// +// Guard rails: only the owning site (by tunnel address) may use a policy; +// destinations on internal address ranges are refused so the edge can't be +// aimed at its own network; and a per-policy rate limit caps how much mail a +// compromised site could push out from the IP. + +// MailHost is somewhere to deliver mail. +type MailHost struct { + Host string + Port int +} + +type smtpErr struct { + Code int + Text string +} + +func (e *smtpErr) Error() string { return strconv.Itoa(e.Code) + " " + e.Text } + +func tempErr(text string) *smtpErr { return &smtpErr{451, "4.4.1 " + text} } + +// asSMTPErr converts an error from net/smtp into a reply for the sender. +func asSMTPErr(err error) *smtpErr { + var te *textproto.Error + if errors.As(err, &te) { + return &smtpErr{te.Code, te.Msg} + } + return tempErr(err.Error()) +} + +func domainOf(addr string) string { + if i := strings.LastIndexByte(addr, '@'); i >= 0 { + return strings.ToLower(strings.Trim(addr[i+1:], "<> ")) + } + return "" +} + +// lookupMail returns where to deliver mail for domain, best first. +func (b *Blacktip) lookupMail(domain string) ([]MailHost, *smtpErr) { + if b.cfg.MailLookup != nil { + hosts, err := b.cfg.MailLookup(domain) + if err != nil { + return nil, tempErr(err.Error()) + } + return hosts, nil + } + mxs, err := net.LookupMX(domain) + var dnsErr *net.DNSError + if err != nil && !(errors.As(err, &dnsErr) && dnsErr.IsNotFound) && len(mxs) == 0 { + return nil, tempErr("DNS lookup for " + domain + " failed: " + err.Error()) + } + if len(mxs) == 1 && mxs[0].Host == "." { + return nil, &smtpErr{556, "5.1.10 " + domain + " does not accept mail"} + } + var out []MailHost + for _, mx := range mxs { + out = append(out, MailHost{Host: strings.TrimSuffix(mx.Host, "."), Port: 25}) + } + if len(out) == 0 { // no MX records: the domain's own address is the mail server (RFC 5321) + if ips, err := net.LookupHost(domain); err != nil || len(ips) == 0 { + return nil, &smtpErr{550, "5.1.2 domain " + domain + " not found"} + } + out = append(out, MailHost{Host: domain, Port: 25}) + } + return out, nil +} + +// mailRateOK counts a new transaction against the policy's per-minute limit. +func (b *Blacktip) mailRateOK(pol model.MailPolicy) bool { + limit := pol.RatePerMinute + if limit <= 0 { + limit = 60 + } + b.mailMu.Lock() + defer b.mailMu.Unlock() + now := time.Now() + kept := b.mailRate[pol.ID][:0] + for _, t := range b.mailRate[pol.ID] { + if now.Sub(t) < time.Minute { + kept = append(kept, t) + } + } + if len(kept) >= limit { + b.mailRate[pol.ID] = kept + return false + } + b.mailRate[pol.ID] = append(kept, now) + return true +} + +type mailSession struct { + b *Blacktip + c net.Conn + pol model.MailPolicy + peer string + cl *smtp.Client + from string + domain string + mx string + rcpts int + dw io.WriteCloser + start time.Time + sts bool // the recipient's MTA-STS policy was enforced + note string // what was done to the message (signed...), for the log +} + +func (b *Blacktip) serveMail(ln net.Listener) { + for { + c, err := ln.Accept() + if err != nil { + return + } + go b.handleMail(c) + } +} + +func (s *mailSession) close() { + if s.cl != nil { + s.cl.Close() + s.cl = nil + } + s.dw, s.rcpts, s.domain = nil, 0, "" +} + +func (s *mailSession) reply(code int, text string) { + b, _ := json.Marshal(tunnel.MailReply{Code: code, Text: text}) + s.c.Write(append(b, '\n')) +} + +func (s *mailSession) replyErr(e *smtpErr) { s.reply(e.Code, e.Text) } + +func (b *Blacktip) handleMail(c net.Conn) { + s := &mailSession{b: b, c: c, peer: hostFromAddr(c.RemoteAddr().String()), start: time.Now()} + defer func() { s.close(); c.Close() }() + br := bufio.NewReader(c) + authed := false + for { + c.SetReadDeadline(time.Now().Add(5 * time.Minute)) + line, err := br.ReadBytes('\n') + if err != nil { + return + } + var m tunnel.MailMsg + if json.Unmarshal(line, &m) != nil { + return + } + if !authed { + b.mu.Lock() + pol, ok := b.mail[m.Egress] + peer := b.peerIP[pol.SiteID] + b.mu.Unlock() + if !ok || peer == "" || peer != s.peer { + s.reply(554, "5.7.1 not allowed: this site may not use that mail egress") + return + } + s.pol, authed = pol, true + } + switch m.Op { + case "rcpt": + s.rcpt(m) + case "data": + if s.cl == nil || s.rcpts == 0 { + s.reply(503, "5.5.1 no recipients") + continue + } + w, err := s.cl.Data() + if err != nil { + s.replyErr(asSMTPErr(err)) + s.close() + continue + } + s.dw = w + s.reply(354, "start mail input") + case "body": + if !s.body(br, m.Len) { + return + } + case "rset", "quit": + if s.cl != nil { + s.cl.Quit() + } + s.close() + s.reply(250, "2.0.0 ok") + if m.Op == "quit" { + return + } + default: + s.reply(500, "5.5.1 unknown request") + } + } +} + +// connect opens a connection to a mail server for domain and gets as far as MAIL FROM. +func (s *mailSession) connect(domain, from string) (*smtp.Client, string, *smtpErr) { + hosts := []MailHost(nil) + if s.pol.Smarthost != "" { + h, p, _ := net.SplitHostPort(s.pol.Smarthost) + port, _ := strconv.Atoi(p) + hosts = []MailHost{{Host: h, Port: port}} + } else { + var e *smtpErr + if hosts, e = s.b.lookupMail(domain); e != nil { + return nil, "", e + } + } + // a recipient domain's MTA-STS policy can forbid some servers and demand verified TLS + requireTLS, why := s.pol.RequireTLS, "this egress requires it" + if s.pol.Smarthost == "" { + if sts := s.b.stsFor(domain); sts != nil && sts.Mode == "enforce" { + var allowed []MailHost + for _, h := range hosts { + if sts.Allows(h.Host) { + allowed = append(allowed, h) + } + } + if len(allowed) == 0 { + return nil, "", &smtpErr{451, "4.7.5 " + domain + "'s MTA-STS policy doesn't allow delivery to any of its listed mail servers"} + } + hosts, requireTLS, why = allowed, true, "the recipient's MTA-STS policy requires it" + s.sts = true + } + } + src := net.ParseIP(s.pol.SourceIP) + last := tempErr("could not reach any mail server for " + domain) + for _, h := range hosts { + var ips []net.IP + if ip := net.ParseIP(h.Host); ip != nil { + ips = []net.IP{ip} + } else { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + res, err := net.DefaultResolver.LookupIPAddr(ctx, h.Host) + cancel() + if err != nil { + last = tempErr("cannot resolve " + h.Host) + continue + } + for _, r := range res { + ips = append(ips, r.IP) + } + } + for _, ip := range ips { + if (src.To4() != nil) != (ip.To4() != nil) { + continue // can't connect to an IPv6 server from an IPv4 source or the reverse + } + if !s.b.cfg.MailAllowPrivate && netguard.Blocked(ip) { + last = &smtpErr{554, "5.7.1 not allowed: " + h.Host + " is an internal address"} + continue + } + d := net.Dialer{Timeout: 15 * time.Second, LocalAddr: &net.TCPAddr{IP: src}} + conn, err := d.Dial("tcp", net.JoinHostPort(ip.String(), strconv.Itoa(h.Port))) + if err != nil { + last = tempErr("cannot connect to " + h.Host + ": " + err.Error()) + continue + } + conn.SetDeadline(time.Now().Add(3 * time.Minute)) + cl, err := smtp.NewClient(conn, h.Host) + if err != nil { + conn.Close() + last = asSMTPErr(err) + continue + } + if err := cl.Hello(s.pol.HeloName); err != nil { + cl.Close() + last = asSMTPErr(err) + continue + } + if ok, _ := cl.Extension("STARTTLS"); ok { + if err := cl.StartTLS(&tls.Config{ServerName: h.Host, InsecureSkipVerify: !requireTLS, RootCAs: s.b.cfg.MailRootCAs}); err != nil { + cl.Close() + last = tempErr("TLS to " + h.Host + " failed: " + err.Error()) + continue + } + } else if requireTLS { + cl.Close() + last = tempErr(h.Host + " does not offer TLS and " + why) + continue + } + if err := cl.Mail(from); err != nil { + cl.Close() + if e := asSMTPErr(err); e.Code >= 500 { + return nil, "", e // the server refused this sender for good: another MX won't differ + } else { + last = e + } + continue + } + return cl, h.Host, nil + } + } + return nil, "", last +} + +func (s *mailSession) rcpt(m tunnel.MailMsg) { + if s.cl == nil { + if !s.b.mailRateOK(s.pol) { + s.reply(451, "4.7.1 sending rate limit reached for this egress, try again shortly") + return + } + domain := domainOf(m.To) + if domain == "" { + s.reply(501, "5.1.3 bad recipient address") + return + } + cl, mx, e := s.connect(domain, m.From) + if e != nil { + s.logEntry(0, e.Code, "mail-failed: "+e.Text, domain, "") + s.replyErr(e) + return + } + s.cl, s.mx, s.domain, s.from = cl, mx, domain, m.From + } else if domainOf(m.To) != s.domain { + s.reply(452, "4.5.3 one destination domain per transaction") + return + } + if err := s.cl.Rcpt(m.To); err != nil { + e := asSMTPErr(err) + if s.rcpts == 0 { + s.close() + } + s.replyErr(e) + return + } + s.rcpts++ + s.reply(250, "2.1.5 ok") +} + +func (s *mailSession) body(br *bufio.Reader, n int) bool { + if n < 0 || n > tunnel.MaxMessage || s.dw == nil || s.cl == nil { + s.reply(552, "5.3.4 message size not accepted") + return false + } + s.c.SetReadDeadline(time.Now().Add(5 * time.Minute)) + data := make([]byte, n) + if _, err := io.ReadFull(br, data); err != nil { + s.close() + return false + } + data, s.note = s.sign(data) + if _, err := s.dw.Write(data); err != nil { + e := asSMTPErr(err) + s.logEntry(n, e.Code, "mail-failed: "+e.Text, s.domain, s.mx) + s.replyErr(e) + s.close() + return true + } + cerr := s.dw.Close() + s.dw = nil + domain, mx := s.domain, s.mx + if cerr != nil { + e := asSMTPErr(cerr) + s.logEntry(n, e.Code, "mail-failed: "+e.Text, domain, mx) + s.replyErr(e) + s.close() + return true + } + s.cl.Quit() + s.logEntry(n, 250, "mail", domain, mx) + s.reply(250, "2.0.0 ok: delivered to "+mx) + s.close() + return true +} + +func (s *mailSession) logEntry(size, code int, result, domain, mx string) { + path := "via " + mx + if s.sts { + path += " (MTA-STS)" + } + if s.note != "" { + path += " · " + s.note + } + s.b.logAccess(model.LogEntry{Time: s.start, Mapping: s.pol.ID, Remote: s.peer, Method: "MAIL", Host: domain, Path: path, + Status: code, Result: result, BytesIn: int64(size), Millis: time.Since(s.start).Milliseconds()}) +} + +// ---------- deliverability checks ---------- + +const defaultMailProbe = "gmail-smtp-in.l.google.com:25" + +// mailCheck inspects how receiving servers will see mail from the policy's source IP. +func (b *Blacktip) mailCheck(pol model.MailPolicy) model.MailCheck { + chk := model.MailCheck{CheckedAt: time.Now(), PTR: []string{}} + src := net.ParseIP(pol.SourceIP) + if names, err := net.LookupAddr(pol.SourceIP); err == nil { + for _, n := range names { + n = strings.ToLower(strings.TrimSuffix(n, ".")) + chk.PTR = append(chk.PTR, n) + if n == strings.ToLower(pol.HeloName) { + chk.PTRMatchesHelo = true + } + } + } + if ips, err := net.LookupHost(pol.HeloName); err == nil { + for _, ip := range ips { + if p := net.ParseIP(ip); p != nil && p.Equal(src) { + chk.HeloResolves = true + } + } + } + if len(pol.DKIM) > 0 { + chk.DKIM = b.checkDKIM(pol.DKIM) + } + probe := b.cfg.MailProbe + if probe == "" { + probe = defaultMailProbe + } + d := net.Dialer{Timeout: 8 * time.Second, LocalAddr: &net.TCPAddr{IP: src}} + c, err := d.Dial("tcp", probe) + if err != nil { + chk.Port25 = "can't connect out on port 25 from this address (" + err.Error() + "); many hosts block it, so ask your provider to open it" + return chk + } + defer c.Close() + c.SetReadDeadline(time.Now().Add(8 * time.Second)) + banner, err := bufio.NewReader(c).ReadString('\n') + if err != nil || !strings.HasPrefix(banner, "220") { + chk.Port25 = "connected, but the server didn't greet us normally" + return chk + } + chk.Port25 = "open" + return chk +} + +// runMailChecks refreshes the report for every current policy. +func (b *Blacktip) runMailChecks() { + b.mu.Lock() + pols := make([]model.MailPolicy, 0, len(b.mail)) + for _, p := range b.mail { + pols = append(pols, p) + } + b.mu.Unlock() + sort.Slice(pols, func(i, j int) bool { return pols[i].ID < pols[j].ID }) + res := map[string]model.MailCheck{} + var wg sync.WaitGroup + var mu sync.Mutex + for _, p := range pols { + wg.Add(1) + go func(p model.MailPolicy) { + defer wg.Done() + c := b.mailCheck(p) + mu.Lock() + res[p.ID] = c + mu.Unlock() + }(p) + } + wg.Wait() + b.mu.Lock() + b.mailChecks = res + b.mu.Unlock() + b.report() +} + +func (b *Blacktip) mailCheckLoop(ctx context.Context) { + t := time.NewTicker(15 * time.Minute) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + b.runMailChecks() + } + } +} + +func mailPolicySig(m map[string]model.MailPolicy) string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + var sb strings.Builder + for _, k := range keys { + p := m[k] + fmt.Fprintf(&sb, "%s|%s|%s|%s;", p.ID, p.SourceIP, p.HeloName, p.Smarthost) + for _, d := range p.DKIM { + fmt.Fprintf(&sb, "%s/%s/%s,", d.KeyID, d.Selector, d.Domain) + } + } + return sb.String() +} diff --git a/mtasts.go b/mtasts.go new file mode 100644 index 0000000..b037a4a --- /dev/null +++ b/mtasts.go @@ -0,0 +1,169 @@ +package blacktip + +import ( + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "git.veillax.com/Veillax/reef/netguard" +) + +// MTA-STS (RFC 8461), sending side. A recipient domain can publish a policy that +// says "mail to us must go to these servers, over verified TLS". Without +// checking it, an attacker who can tamper with DNS or the network can redirect +// or downgrade your outgoing mail. When a policy says "enforce" we only deliver +// to the listed servers and only over TLS with a valid certificate. + +// STSPolicy is a recipient domain's MTA-STS policy. +type STSPolicy struct { + Mode string // "enforce", "testing" or "none" + MX []string // exact host names or "*.example.com" patterns + MaxAge time.Duration +} + +// Allows reports whether host is a mail server the policy permits. +func (p *STSPolicy) Allows(host string) bool { + host = strings.ToLower(strings.TrimSuffix(host, ".")) + for _, pat := range p.MX { + pat = strings.ToLower(strings.TrimSuffix(pat, ".")) + if strings.HasPrefix(pat, "*.") { + // a wildcard stands for exactly one label + if rest, ok := strings.CutSuffix(host, pat[1:]); ok && rest != "" && !strings.Contains(rest, ".") { + return true + } + } else if pat == host { + return true + } + } + return false +} + +func parseSTS(text string) (*STSPolicy, error) { + p := &STSPolicy{} + version := "" + for _, line := range strings.Split(strings.ReplaceAll(text, "\r\n", "\n"), "\n") { + k, v, ok := strings.Cut(line, ":") + if !ok { + continue + } + k, v = strings.ToLower(strings.TrimSpace(k)), strings.TrimSpace(v) + switch k { + case "version": + version = v + case "mode": + p.Mode = strings.ToLower(v) + case "mx": + p.MX = append(p.MX, v) + case "max_age": + n, err := strconv.ParseInt(v, 10, 64) + if err != nil || n < 0 { + return nil, errors.New("bad max_age") + } + if n > 31557600 { + n = 31557600 // a year is the most the RFC allows + } + p.MaxAge = time.Duration(n) * time.Second + } + } + switch { + case version != "STSv1": + return nil, errors.New("not an STSv1 policy") + case p.Mode != "enforce" && p.Mode != "testing" && p.Mode != "none": + return nil, fmt.Errorf("unknown mode %q", p.Mode) + case p.Mode != "none" && len(p.MX) == 0: + return nil, errors.New("policy lists no mx") + case p.MaxAge <= 0: + return nil, errors.New("policy has no max_age") + } + return p, nil +} + +type stsEntry struct { + id string + pol *STSPolicy + expires time.Time +} + +type stsCache struct { + mu sync.Mutex + m map[string]*stsEntry +} + +func (b *Blacktip) lookupTXT(name string) ([]string, error) { + if b.cfg.MailLookupTXT != nil { + return b.cfg.MailLookupTXT(name) + } + return netLookupTXT(name) +} + +// fetchSTS downloads and parses the policy file. The recipient domain is +// attacker-controlled, so the request goes through the guarded client: no +// internal addresses, no redirects. +func fetchSTS(client *http.Client, domain string) (*STSPolicy, error) { + resp, err := client.Get("https://mta-sts." + domain + "/.well-known/mta-sts.txt") + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return nil, fmt.Errorf("policy server answered %s", resp.Status) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) + if err != nil { + return nil, err + } + return parseSTS(string(body)) +} + +// stsFor returns the domain's current MTA-STS policy, or nil if it has none. +func (b *Blacktip) stsFor(domain string) *STSPolicy { + if b.cfg.MailSTS != nil { + p, _ := b.cfg.MailSTS(domain) + return p + } + id := "" + if txts, err := b.lookupTXT("_mta-sts." + domain); err == nil { + for _, t := range txts { + if strings.HasPrefix(strings.TrimSpace(t), "v=STSv1") { + for _, f := range strings.Split(t, ";") { + if k, v, ok := strings.Cut(strings.TrimSpace(f), "="); ok && k == "id" { + id = v + } + } + } + } + } + now := time.Now() + b.sts.mu.Lock() + if b.sts.m == nil { + b.sts.m = map[string]*stsEntry{} + } + ent := b.sts.m[domain] + b.sts.mu.Unlock() + valid := ent != nil && now.Before(ent.expires) + if id == "" { + if valid { + return ent.pol // a cached policy stays in force until it expires + } + return nil + } + if valid && ent.id == id { + return ent.pol + } + pol, err := fetchSTS(netguard.Client(b.cfg.MailAllowPrivate), domain) + if err != nil { + if valid { + return ent.pol + } + return nil + } + b.sts.mu.Lock() + b.sts.m[domain] = &stsEntry{id: id, pol: pol, expires: now.Add(pol.MaxAge)} + b.sts.mu.Unlock() + return pol +} diff --git a/mtasts_test.go b/mtasts_test.go new file mode 100644 index 0000000..5749185 --- /dev/null +++ b/mtasts_test.go @@ -0,0 +1,107 @@ +package blacktip + +import ( + "context" + "crypto/tls" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestParseSTS(t *testing.T) { + good := "version: STSv1\r\nmode: enforce\r\nmx: mail.example.com\r\nmx: *.mail.example.com\r\nmax_age: 86400\r\n" + p, err := parseSTS(good) + if err != nil || p.Mode != "enforce" || len(p.MX) != 2 || p.MaxAge != 24*time.Hour { + t.Fatalf("parse: %+v %v", p, err) + } + if p, _ := parseSTS(strings.ReplaceAll(good, "86400", "999999999999")); p == nil || p.MaxAge != 31557600*time.Second { + t.Errorf("max_age must be capped at a year: %+v", p) + } + if p, err := parseSTS("version: STSv1\nmode: none\nmax_age: 60\n"); err != nil || p.Mode != "none" { + t.Errorf("mode none needs no mx: %+v %v", p, err) + } + for name, bad := range map[string]string{ + "wrong version": "version: STSv2\nmode: enforce\nmx: a.b\nmax_age: 1\n", + "unknown mode": "version: STSv1\nmode: maybe\nmx: a.b\nmax_age: 1\n", + "no mx": "version: STSv1\nmode: enforce\nmax_age: 1\n", + "no max_age": "version: STSv1\nmode: enforce\nmx: a.b\n", + "bad max_age": "version: STSv1\nmode: enforce\nmx: a.b\nmax_age: soon\n", + "empty": "", + } { + if _, err := parseSTS(bad); err == nil { + t.Errorf("accepted a policy with %s", name) + } + } +} + +func TestSTSMXMatching(t *testing.T) { + p := &STSPolicy{MX: []string{"mail.example.com", "*.mx.example.com"}} + for host, want := range map[string]bool{ + "mail.example.com": true, + "MAIL.Example.com.": true, // case and a trailing dot don't matter + "a.mx.example.com": true, + "a.b.mx.example.com": false, // a wildcard is exactly one label + "mx.example.com": false, // and never zero labels + "evil-mail.example.com": false, + "mail.example.com.evil": false, + "": false, + } { + if got := p.Allows(host); got != want { + t.Errorf("Allows(%q) = %v, want %v", host, got, want) + } + } +} + +func TestFetchSTSHandlesTheHTTPSExchange(t *testing.T) { + var gotHost, gotPath string + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHost, gotPath = r.Host, r.URL.Path + switch r.Header.Get("X-Case") { + case "redirect": + http.Redirect(w, r, "https://elsewhere.invalid/", http.StatusFound) + case "missing": + http.NotFound(w, r) + default: + w.Write([]byte("version: STSv1\nmode: testing\nmx: mx.example.test\nmax_age: 600\n")) + } + })) + defer srv.Close() + // a client that sends every request to the test server regardless of the name + client := func(c string) *http.Client { + return &http.Client{ + Transport: &roundTripper{c: c, rt: &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DialContext: func(ctx context.Context, n, a string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, n, strings.TrimPrefix(srv.URL, "https://")) + }}}, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + } + } + p, err := fetchSTS(client(""), "example.test") + if err != nil || p.Mode != "testing" || !p.Allows("mx.example.test") { + t.Fatalf("fetch: %+v %v", p, err) + } + if gotHost != "mta-sts.example.test" || gotPath != "/.well-known/mta-sts.txt" { + t.Errorf("fetched %s%s, want mta-sts./.well-known/mta-sts.txt", gotHost, gotPath) + } + if _, err := fetchSTS(client("missing"), "example.test"); err == nil { + t.Error("a 404 policy must be an error") + } + if _, err := fetchSTS(client("redirect"), "example.test"); err == nil { + t.Error("a redirect must not be followed or accepted") + } +} + +type roundTripper struct { + c string + rt http.RoundTripper +} + +func (r *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + req.Header.Set("X-Case", r.c) + return r.rt.RoundTrip(req) +} diff --git a/proxyproto.go b/proxyproto.go new file mode 100644 index 0000000..454845c --- /dev/null +++ b/proxyproto.go @@ -0,0 +1,65 @@ +package blacktip + +import ( + "encoding/binary" + "fmt" + "net" + "net/netip" +) + +// PROXY protocol (HAProxy spec): a short header telling the target who the real +// client was. Without it a service behind the tunnel only ever sees the tunnel, +// so IP bans, rate limits, spam checks and logs all see the wrong address. +// Servers must be told to expect it and to trust only the edge/site that sends it. + +var proxyV2Sig = []byte("\r\n\r\n\x00\r\nQUIT\n") + +func addrPort(a net.Addr) (netip.Addr, uint16, bool) { + ta, ok := a.(*net.TCPAddr) + if !ok { + ap, err := netip.ParseAddrPort(a.String()) + if err != nil { + return netip.Addr{}, 0, false + } + return ap.Addr().Unmap(), ap.Port(), true + } + ip, ok := netip.AddrFromSlice(ta.IP) + if !ok { + return netip.Addr{}, 0, false + } + return ip.Unmap(), uint16(ta.Port), true +} + +// proxyHeader builds a PROXY protocol header ("v1" or "v2") for a TCP connection from src to dst. +func proxyHeader(version string, src, dst net.Addr) []byte { + sip, sport, ok1 := addrPort(src) + dip, dport, ok2 := addrPort(dst) + known := ok1 && ok2 && sip.Is4() == dip.Is4() + if version == "v2" { + h := append([]byte{}, proxyV2Sig...) + if !known { + return append(h, 0x20, 0x00, 0x00, 0x00) // LOCAL: no address information + } + if sip.Is4() { + h = append(h, 0x21, 0x11, 0x00, 12) + s4, d4 := sip.As4(), dip.As4() + h = append(h, s4[:]...) + h = append(h, d4[:]...) + } else { + h = append(h, 0x21, 0x21, 0x00, 36) + s6, d6 := sip.As16(), dip.As16() + h = append(h, s6[:]...) + h = append(h, d6[:]...) + } + h = binary.BigEndian.AppendUint16(h, sport) + return binary.BigEndian.AppendUint16(h, dport) + } + if !known { + return []byte("PROXY UNKNOWN\r\n") + } + fam := "TCP4" + if !sip.Is4() { + fam = "TCP6" + } + return []byte(fmt.Sprintf("PROXY %s %s %s %d %d\r\n", fam, sip, dip, sport, dport)) +} diff --git a/traefik.go b/traefik.go new file mode 100644 index 0000000..be9c05f --- /dev/null +++ b/traefik.go @@ -0,0 +1,600 @@ +package blacktip + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "log" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.veillax.com/Veillax/reef/model" + "git.veillax.com/Veillax/reef/tunnel" +) + +// TraefikConfig turns on Traefik as the HTTP/HTTPS front end. +// +// Traefik declares its listening ports in static configuration, so it can only +// own the ports for web resources (a handful of ip:443/80 pairs that rarely +// change). Blacktip writes that file, restarts Traefik only when the set of web +// listeners changes, and hot-reloads everything else (routes, logins, targets) +// through Traefik's file provider. Raw TCP, UDP and TLS-passthrough ports stay +// with Blacktip's own listeners, which open instantly with no restart. +type TraefikConfig struct { + Path string // path to the traefik binary; empty = use the built-in proxy + ACMEDNS string // DNS-challenge provider (e.g. "cloudflare") for wildcard certificates + NoACME bool // serve Traefik's built-in self-signed certificate instead of Let's Encrypt +} + +type traefikMgr struct { + b *Blacktip + cfg TraefikConfig + dir string + gatePort int + + mu sync.Mutex + staticSig string + stop context.CancelFunc // stops the current traefik process + lastErr string + maps map[string]*mapState + ports map[string]int // mapping ID -> relay port, kept stable across edits so Traefik's backend URL never changes + dynBytes []byte +} + +func newTraefik(b *Blacktip, cfg TraefikConfig) (*traefikMgr, error) { + if _, err := exec.LookPath(cfg.Path); err != nil { + if _, err2 := os.Stat(cfg.Path); err2 != nil { + return nil, fmt.Errorf("traefik binary %q not found: %w", cfg.Path, err) + } + } + t := &traefikMgr{b: b, cfg: cfg, dir: filepath.Join(b.cfg.DataDir, "traefik"), maps: map[string]*mapState{}, ports: map[string]int{}} + if err := os.MkdirAll(t.dir, 0o700); err != nil { + return nil, err + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return nil, err + } + t.gatePort = ln.Addr().(*net.TCPAddr).Port + mux := http.NewServeMux() + mux.HandleFunc("/auth/", t.forwardAuth) + mux.HandleFunc("/.hammerhead/", t.gateCallback) + go (&http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second}).Serve(ln) + return t, nil +} + +func (t *traefikMgr) run(ctx context.Context) { + go t.tailAccessLog(ctx) + go t.probeCerts(ctx) + <-ctx.Done() + t.mu.Lock() + if t.stop != nil { + t.stop() + } + t.mu.Unlock() +} + +// ---------- the gate: forward-auth and login callbacks ---------- + +func (t *traefikMgr) mapByID(id string) *mapState { + t.mu.Lock() + defer t.mu.Unlock() + return t.maps[id] +} + +func firstIP(xff string) string { + if i := strings.IndexByte(xff, ','); i >= 0 { + xff = xff[:i] + } + return strings.TrimSpace(xff) +} + +// forwardAuth is called by Traefik before every request to a protected resource. +func (t *traefikMgr) forwardAuth(w http.ResponseWriter, r *http.Request) { + ms := t.mapByID(strings.TrimPrefix(r.URL.Path, "/auth/")) + if ms == nil { + http.Error(w, "unknown resource", http.StatusForbidden) + return + } + in := authInput{ + Method: r.Header.Get("X-Forwarded-Method"), Scheme: r.Header.Get("X-Forwarded-Proto"), Host: r.Header.Get("X-Forwarded-Host"), + URI: r.Header.Get("X-Forwarded-Uri"), ClientIP: firstIP(r.Header.Get("X-Forwarded-For")), Cookie: r.Header.Get("Cookie"), + Upgrade: strings.EqualFold(r.Header.Get("Upgrade"), "websocket"), + } + d := t.b.decide(ms, in) + if d.Result != "allowed" && d.Result != "bypass" { + t.b.logAccess(model.LogEntry{Time: time.Now(), Mapping: ms.m.ID, Remote: in.ClientIP, Method: in.Method, Host: hostOnly(in.Host), Path: in.URI, Result: d.Result, Status: d.Status}) + } + if !d.Allow { + if d.Location != "" { + w.Header().Set("Location", d.Location) + } + w.WriteHeader(d.Status) + return + } + if d.User != "" { + w.Header().Set("X-Hammerhead-User", d.User) + } + if d.Cookie != "" { + w.Header().Set("Cookie", d.Cookie) + } + w.WriteHeader(http.StatusOK) +} + +// gateCallback serves /.hammerhead/callback and /.hammerhead/logout. Traefik routes +// those paths here and a middleware stamps which mapping the request is for +// (Traefik ignores paths in backend URLs, so a header is the reliable way). +func (t *traefikMgr) gateCallback(w http.ResponseWriter, r *http.Request) { + id := r.Header.Get("X-Hammerhead-Mapping") + ms := t.mapByID(id) + if ms == nil { + http.NotFound(w, r) + return + } + secure := r.Header.Get("X-Forwarded-Proto") == "https" + switch r.URL.Path { + case "/.hammerhead/callback": + user, result := t.b.callback(w, r, ms, secure) + t.b.logAccess(model.LogEntry{Time: time.Now(), Mapping: id, Remote: firstIP(r.Header.Get("X-Forwarded-For")), Host: hostOnly(r.Header.Get("X-Forwarded-Host")), Path: "/.hammerhead/callback", User: user, Result: result}) + case "/.hammerhead/logout": + t.b.logout(w, r, secure) + default: + http.NotFound(w, r) + } +} + +// ---------- relays: loopback ports Traefik uses to reach targets through the tunnel ---------- + +func (ms *mapState) startRelay() { + var ln net.Listener + var err error + if ms.b.tr != nil { + ms.b.tr.mu.Lock() + pref := ms.b.tr.ports[ms.m.ID] + ms.b.tr.mu.Unlock() + if pref != 0 { + ln, err = net.Listen("tcp", "127.0.0.1:"+strconv.Itoa(pref)) + } + } + if ln == nil { + ln, err = net.Listen("tcp", "127.0.0.1:0") + } + if err != nil { + log.Printf("relay for %s: %v", ms.m.Name, err) + return + } + ms.relayLn = ln + ms.relayPort = ln.Addr().(*net.TCPAddr).Port + if ms.b.tr != nil { + ms.b.tr.mu.Lock() + ms.b.tr.ports[ms.m.ID] = ms.relayPort + ms.b.tr.mu.Unlock() + } + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { + up, _, err := ms.dial("tcp", -1) + if err != nil { + c.Close() + return + } + tunnel.Proxy(c, c, up, up) + }() + } + }() +} + +// ---------- configuration ---------- + +var nonAlnum = regexp.MustCompile(`[^a-zA-Z0-9]+`) + +func epName(ip string, port int) string { + return "ep-" + nonAlnum.ReplaceAllString(ip, "_") + "-" + strconv.Itoa(port) +} + +func hostRule(host string) (string, bool) { + if strings.HasPrefix(host, "*.") { + return "HostRegexp(`^[^.]+\\." + regexp.QuoteMeta(host[2:]) + "$`)", true + } + return "Host(`" + host + "`)", false +} + +// sync rewrites Traefik's configuration for the given web mappings and +// (re)starts the process if its static part changed. Called with b.mu held. +func (t *traefikMgr) sync(ctx context.Context, maps []*mapState, acmeEmail string, staging bool) { + t.mu.Lock() + defer t.mu.Unlock() + t.maps = map[string]*mapState{} + eps := map[string]string{} // name -> address + needACME := false + for _, ms := range maps { + t.maps[ms.m.ID] = ms + eps[epName(ms.m.BindIP, ms.m.Port)] = net.JoinHostPort(ms.m.BindIP, strconv.Itoa(ms.m.Port)) + if ms.m.Proto == model.ProtoHTTPS { + needACME = true + if ms.m.RedirectHTTP { + eps[epName(ms.m.BindIP, 80)] = net.JoinHostPort(ms.m.BindIP, "80") + } + } + } + routers := map[string]any{} + services := map[string]any{} + mws := map[string]any{ + "strip-user": map[string]any{"headers": map[string]any{"customRequestHeaders": map[string]string{"X-Hammerhead-User": ""}}}, + } + for _, ms := range maps { + m := ms.m + if ms.relayPort == 0 { + continue + } + rule, wildcard := hostRule(m.Hostname) + ep := epName(m.BindIP, m.Port) + prio := 5000 + if wildcard { + prio = 2000 + } + scheme := "http" + if m.BackendHTTPS { + scheme = "https" + } + services["s-"+m.ID] = map[string]any{"loadBalancer": map[string]any{ + "servers": []map[string]string{{"url": fmt.Sprintf("%s://127.0.0.1:%d", scheme, ms.relayPort)}}, + "passHostHeader": true, + "serversTransport": "insecure", + }} + chain := []string{} + if m.Auth.Protected() { + mws["auth-"+m.ID] = map[string]any{"forwardAuth": map[string]any{ + "address": fmt.Sprintf("http://127.0.0.1:%d/auth/%s", t.gatePort, m.ID), + "authResponseHeaders": []string{"X-Hammerhead-User", "Cookie"}, + "trustForwardHeader": true, + "maxResponseBodySize": 8192, + }} + chain = append(chain, "auth-"+m.ID) + } else { + chain = append(chain, "strip-user") + } + if m.RewriteHost != "" { + mws["host-"+m.ID] = map[string]any{"headers": map[string]any{"customRequestHeaders": map[string]string{"Host": m.RewriteHost}}} + chain = append(chain, "host-"+m.ID) + } + router := map[string]any{"rule": rule, "entryPoints": []string{ep}, "service": "s-" + m.ID, "middlewares": chain, "priority": prio} + var tlsCfg map[string]any + if m.Proto == model.ProtoHTTPS { + tlsCfg = map[string]any{} + if !t.cfg.NoACME { + tlsCfg["certResolver"] = "le" + if wildcard { + tlsCfg["domains"] = []map[string]any{{"main": m.Hostname}} + } + } + router["tls"] = tlsCfg + } + routers["m-"+m.ID] = router + + if m.Auth.Protected() { // login callbacks are answered by the gate, ahead of the app + services["g-"+m.ID] = map[string]any{"loadBalancer": map[string]any{ + "servers": []map[string]string{{"url": fmt.Sprintf("http://127.0.0.1:%d", t.gatePort)}}, "passHostHeader": true}} + mws["gate-"+m.ID] = map[string]any{"headers": map[string]any{"customRequestHeaders": map[string]string{"X-Hammerhead-Mapping": m.ID}}} + g := map[string]any{"rule": rule + " && PathPrefix(`/.hammerhead/`)", "entryPoints": []string{ep}, "service": "g-" + m.ID, + "middlewares": []string{"gate-" + m.ID}, "priority": prio + 1000} + if tlsCfg != nil { + g["tls"] = tlsCfg + } + routers["g-"+m.ID] = g + } + if m.Proto == model.ProtoHTTPS && m.RedirectHTTP { + redirPort := "" + if m.Port != 443 { + redirPort = strconv.Itoa(m.Port) + } + mws["redir-"+m.ID] = map[string]any{"redirectScheme": map[string]any{"scheme": "https", "port": redirPort, "permanent": true}} + routers["r-"+m.ID] = map[string]any{"rule": rule, "entryPoints": []string{epName(m.BindIP, 80)}, "service": "noop@internal", + "middlewares": []string{"redir-" + m.ID}, "priority": prio} + } + } + dynamic := map[string]any{"http": map[string]any{ + "routers": routers, "services": services, "middlewares": mws, + "serversTransports": map[string]any{"insecure": map[string]any{"insecureSkipVerify": true}}, + }} + t.writeDynamic(dynamic) // JSON is valid YAML + + // static part: a change here needs a restart + names := make([]string, 0, len(eps)) + for n := range eps { + names = append(names, n) + } + sort.Strings(names) + entry := map[string]any{} + for _, n := range names { + // "delete" drops header aliases such as X_Hammerhead_User, which CGI/PHP style + // backends would otherwise read as the trusted X-Hammerhead-User + entry[n] = map[string]any{"address": eps[n], "http": map[string]any{"aliasHeadersStrategy": "delete"}} + } + static := map[string]any{ + "global": map[string]any{"checkNewVersion": false, "sendAnonymousUsage": false}, + "log": map[string]any{"level": "INFO"}, + "accessLog": map[string]any{"filePath": filepath.Join(t.dir, "access.log"), "format": "json", "fields": map[string]any{ + "defaultMode": "drop", + "names": map[string]string{"StartUTC": "keep", "RouterName": "keep", "ClientHost": "keep", "RequestMethod": "keep", "RequestHost": "keep", + "RequestPath": "keep", "DownstreamStatus": "keep", "DownstreamContentSize": "keep", "RequestContentSize": "keep", "Duration": "keep"}}}, + "entryPoints": entry, + "providers": map[string]any{"file": map[string]any{"filename": filepath.Join(t.dir, "dynamic.yml"), "watch": true}}, + } + if needACME && !t.cfg.NoACME { + acme := map[string]any{"email": acmeEmail, "storage": filepath.Join(t.dir, "acme.json")} + if staging { + acme["caServer"] = "https://acme-staging-v02.api.letsencrypt.org/directory" + } + if t.cfg.ACMEDNS != "" { + acme["dnsChallenge"] = map[string]any{"provider": t.cfg.ACMEDNS} + } else { + acme["tlsChallenge"] = map[string]any{} + } + static["certificatesResolvers"] = map[string]any{"le": map[string]any{"acme": acme}} + } + sb, _ := json.Marshal(static) + sig := string(sb) + if len(eps) == 0 { // nothing to serve: stop Traefik until a web port exists + if t.stop != nil { + t.stop() + t.stop = nil + } + t.staticSig = "" + return + } + if sig == t.staticSig && t.stop != nil { + return + } + if t.stop != nil { + t.stop() + } + writeJSONAtomic(filepath.Join(t.dir, "static.yml"), static) + t.staticSig = sig + pctx, cancel := context.WithCancel(ctx) + t.stop = cancel + go t.supervise(pctx) +} + +// writeDynamic updates Traefik's dynamic configuration in place. File watchers +// on some platforms miss atomic renames, so the same bytes are written again +// shortly after: a missed event can never leave stale routes behind. +func (t *traefikMgr) writeDynamic(v any) { + b, _ := json.MarshalIndent(v, "", " ") + path := filepath.Join(t.dir, "dynamic.yml") + t.dynBytes = b + os.WriteFile(path, b, 0o600) + for _, d := range []time.Duration{time.Second, 4 * time.Second} { + time.AfterFunc(d, func() { + t.mu.Lock() + defer t.mu.Unlock() + if string(t.dynBytes) == string(b) { // still the latest + os.WriteFile(path, b, 0o600) + } + }) + } +} + +func writeJSONAtomic(path string, v any) { + b, _ := json.MarshalIndent(v, "", " ") + tmp := path + ".tmp" + if os.WriteFile(tmp, b, 0o600) == nil { + os.Rename(tmp, path) + } +} + +// ---------- process supervision ---------- + +type lineLogger struct{ prefix string } + +func (l lineLogger) Write(p []byte) (int, error) { + sc := bufio.NewScanner(strings.NewReader(string(p))) + for sc.Scan() { + log.Printf("%s%s", l.prefix, sc.Text()) + } + return len(p), nil +} + +func (t *traefikMgr) supervise(ctx context.Context) { + for ctx.Err() == nil { + cmd := exec.CommandContext(ctx, t.cfg.Path, "--configFile="+filepath.Join(t.dir, "static.yml")) + cmd.Dir = t.dir + cmd.Stdout, cmd.Stderr = lineLogger{"traefik: "}, lineLogger{"traefik: "} + err := cmd.Start() + if err == nil { + t.setErr("") + log.Printf("traefik started (pid %d)", cmd.Process.Pid) + err = cmd.Wait() + } + if ctx.Err() != nil { + return + } + t.setErr(fmt.Sprintf("traefik stopped: %v", err)) + log.Printf("traefik stopped: %v; restarting in 3s", err) + select { + case <-time.After(3 * time.Second): + case <-ctx.Done(): + } + } +} + +func (t *traefikMgr) setErr(s string) { + t.mu.Lock() + t.lastErr = s + t.mu.Unlock() +} + +func (t *traefikMgr) err() string { + t.mu.Lock() + defer t.mu.Unlock() + return t.lastErr +} + +// ---------- access log import ---------- + +type traefikLine struct { + StartUTC time.Time `json:"StartUTC"` + RouterName string `json:"RouterName"` + ClientHost string `json:"ClientHost"` + RequestMethod string `json:"RequestMethod"` + RequestHost string `json:"RequestHost"` + RequestPath string `json:"RequestPath"` + DownstreamStatus int `json:"DownstreamStatus"` + DownstreamSize int64 `json:"DownstreamContentSize"` + RequestSize int64 `json:"RequestContentSize"` + DurationNanos int64 `json:"Duration"` +} + +// tailAccessLog turns Traefik's JSON access log into Blacktip's log entries and traffic counters. +// Lines older than this process are ignored (by timestamp, so requests served +// before the first poll still count). +func (t *traefikMgr) tailAccessLog(ctx context.Context) { + path := filepath.Join(t.dir, "access.log") + startedAt := time.Now().Add(-2 * time.Second) + var off int64 + first := true + tick := time.NewTicker(time.Second) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + } + f, err := os.Open(path) + if err != nil { + continue + } + st, _ := f.Stat() + switch { + case first && st.Size() > 4<<20: + off = st.Size() - 4<<20 // only the recent tail of an old log + case st.Size() < off: + off = 0 // truncated + } + first = false + f.Seek(off, io.SeekStart) + r := bufio.NewReader(f) + for { + line, err := r.ReadBytes('\n') + if err != nil { + break // partial line: read it next time + } + off += int64(len(line)) + var tl traefikLine + if json.Unmarshal(line, &tl) != nil || !strings.HasPrefix(tl.RouterName, "m-") || tl.StartUTC.Before(startedAt) { + continue + } + id := strings.TrimSuffix(strings.TrimPrefix(tl.RouterName, "m-"), "@file") + ms := t.mapByID(id) + if ms == nil { + continue + } + ms.conns.Add(1) + ms.bytesIn.Add(tl.RequestSize) + ms.bytesOut.Add(tl.DownstreamSize) + t.b.logAccess(model.LogEntry{Time: tl.StartUTC, Mapping: id, Remote: tl.ClientHost, Method: tl.RequestMethod, Host: hostOnly(tl.RequestHost), + Path: tl.RequestPath, Status: tl.DownstreamStatus, Result: "allowed", BytesIn: tl.RequestSize, BytesOut: tl.DownstreamSize, Millis: tl.DurationNanos / 1e6}) + } + f.Close() + if st.Size() > 64<<20 { // keep the log from growing forever (best effort) + if os.Truncate(path, 0) == nil { + off = 0 + } + } + } +} + +// ---------- certificate status ---------- + +// probeCerts handshakes with Traefik to see which certificate it serves for each https hostname. +func (t *traefikMgr) probeCerts(ctx context.Context) { + tick := time.NewTicker(30 * time.Second) + defer tick.Stop() + first := time.After(4 * time.Second) + for { + select { + case <-ctx.Done(): + return + case <-first: + case <-tick.C: + } + t.mu.Lock() + var targets []*mapState + for _, ms := range t.maps { + if ms.m.Proto == model.ProtoHTTPS && !strings.HasPrefix(ms.m.Hostname, "*") { + targets = append(targets, ms) + } + } + t.mu.Unlock() + changed := false + for _, ms := range targets { + ip := ms.m.BindIP + if pip := net.ParseIP(ip); pip != nil && pip.IsUnspecified() { + ip = "127.0.0.1" + } + status := t.probeOne(net.JoinHostPort(ip, strconv.Itoa(ms.m.Port)), ms.m.Hostname) + t.b.mu.Lock() + if t.b.acme.certs == nil { + t.b.acme.certs = map[string]string{} + } + if t.b.acme.certs[ms.m.Hostname] != status { + t.b.acme.certs[ms.m.Hostname] = status + changed = true + } + t.b.mu.Unlock() + } + if changed { + go t.b.report() + } + } +} + +func (t *traefikMgr) probeOne(addr, host string) string { + d := &net.Dialer{Timeout: 3 * time.Second} + c, err := tls.DialWithDialer(d, "tcp", addr, &tls.Config{ServerName: host, InsecureSkipVerify: true}) + if err != nil { + return "error: " + err.Error() + } + defer c.Close() + certs := c.ConnectionState().PeerCertificates + if len(certs) == 0 { + return "error: no certificate" + } + leaf := certs[0] + switch { + case leaf.Subject.CommonName == "TRAEFIK DEFAULT CERT": + if t.cfg.NoACME { + return "self-signed (Traefik default certificate)" + } + return "pending (still serving Traefik's default certificate)" + case leaf.Subject.CommonName == host || containsStr(leaf.DNSNames, host): + return "valid until " + leaf.NotAfter.Format("2006-01-02") + " (" + strings.Join(leaf.Issuer.Organization, ", ") + ")" + } + return "error: certificate is for " + leaf.Subject.CommonName +} + +func containsStr(list []string, s string) bool { + for _, x := range list { + if x == s { + return true + } + } + return false +} diff --git a/web.go b/web.go new file mode 100644 index 0000000..1909791 --- /dev/null +++ b/web.go @@ -0,0 +1,310 @@ +package blacktip + +import ( + "context" + "crypto/tls" + "fmt" + "io" + "log" + "net" + "net/http" + "net/http/httputil" + "net/url" + "strconv" + "strings" + "time" + + "git.veillax.com/Veillax/reef/authtoken" + "git.veillax.com/Veillax/reef/model" +) + +const authCookie = "hh_auth" + +// recorder captures what a handler sent, without breaking websockets or +// flushing (http.ResponseController follows Unwrap). +type recorder struct { + http.ResponseWriter + status int + bytes int64 +} + +func (r *recorder) WriteHeader(code int) { + if r.status == 0 { + r.status = code + } + r.ResponseWriter.WriteHeader(code) +} + +func (r *recorder) Write(p []byte) (int, error) { + if r.status == 0 { + r.status = 200 + } + n, err := r.ResponseWriter.Write(p) + r.bytes += int64(n) + return n, err +} + +func (r *recorder) Unwrap() http.ResponseWriter { return r.ResponseWriter } + +// initProxy builds the built-in reverse proxy (used when Traefik isn't). +func (ms *mapState) initProxy() { + scheme, port := "http", 80 + if ms.m.BackendHTTPS { + scheme, port = "https", 443 + } + ms.transport = &http.Transport{ + // The request URL host is "t": it names the target and keeps + // pooled connections to different targets apart. + DialContext: func(ctx context.Context, _, addr string) (net.Conn, error) { + idx := -1 + if h, _, err := net.SplitHostPort(addr); err == nil && strings.HasPrefix(h, "t") { + if n, err := strconv.Atoi(h[1:]); err == nil { + idx = n + } + } + c, _, err := ms.dial("tcp", idx) + return c, err + }, + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // backends are usually self-signed + TLSHandshakeTimeout: 10 * time.Second, + MaxIdleConnsPerHost: 32, + IdleConnTimeout: 60 * time.Second, + } + ms.proxy = &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetXForwarded() + idx := ms.order()[0] + pr.Out.URL.Scheme = scheme + pr.Out.URL.Host = fmt.Sprintf("t%d:%d", idx, port) + if ms.m.RewriteHost != "" { + pr.Out.Host = ms.m.RewriteHost + } + }, + Transport: ms.transport, + FlushInterval: -1, + ErrorLog: log.New(io.Discard, "", 0), + ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { + http.Error(w, "Bad gateway: the service behind this address isn't answering.", http.StatusBadGateway) + }, + } +} + +func ruleMatches(ru model.Rule, path, clientIP string) bool { + switch ru.Match { + case "path": + return strings.HasPrefix(path, ru.Value) + case "ip": + ip := net.ParseIP(clientIP) + if ip == nil { + return false + } + if p := net.ParseIP(ru.Value); p != nil { + return p.Equal(ip) + } + if _, n, err := net.ParseCIDR(ru.Value); err == nil { + return n.Contains(ip) + } + } + return false +} + +func safeDest(p string) string { + if p == "" || p[0] != '/' || strings.HasPrefix(p, "//") || strings.HasPrefix(p, "/\\") { + return "/" + } + return p +} + +func hostFromAddr(a string) string { + if h, _, err := net.SplitHostPort(a); err == nil { + return h + } + return a +} + +// ---------- the access decision, shared by the built-in proxy and the Traefik gate ---------- + +type authInput struct { + Method, Scheme, Host, URI, ClientIP, Cookie string + Upgrade bool +} + +type authDecision struct { + Allow bool + Status int // when !Allow: 302, 401 or 403 + Location string // for 302 + User string + Cookie string // the request's Cookie header with our own cookie removed + Result string +} + +func cookieValue(header, name string) string { + for _, c := range (&http.Request{Header: http.Header{"Cookie": {header}}}).Cookies() { + if c.Name == name { + return c.Value + } + } + return "" +} + +func stripCookie(header string) string { + var keep []string + for _, c := range (&http.Request{Header: http.Header{"Cookie": {header}}}).Cookies() { + if c.Name != authCookie { + keep = append(keep, c.Name+"="+c.Value) + } + } + return strings.Join(keep, "; ") +} + +// decide applies a web mapping's rules and login policy to one request. +func (b *Blacktip) decide(ms *mapState, in authInput) authDecision { + m := ms.m + path, query, rawPath := "/", url.Values{}, "/" + if u, err := url.ParseRequestURI(in.URI); err == nil { + path, rawPath, query = u.Path, u.EscapedPath(), u.Query() + } + d := authDecision{Cookie: stripCookie(in.Cookie), Result: "allowed"} + bypass := false + for _, ru := range m.Rules { + if !ruleMatches(ru, path, in.ClientIP) { + continue + } + if ru.Action == "deny" { + return authDecision{Status: http.StatusForbidden, Result: "denied"} + } + bypass, d.Result = true, "bypass" + break + } + if !m.Auth.Protected() || bypass { + d.Allow = true + return d + } + b.mu.Lock() + key, dash := b.authKey, b.dashboard + b.mu.Unlock() + if t, ok := authtoken.Verify(key, cookieValue(in.Cookie, authCookie), authtoken.KindSession, m.ID, ms.epoch.Load()); ok { + d.Allow, d.User = true, t.Sub + return d + } + share := query.Get("hh_share") + query.Del("hh_share") + back := in.Scheme + "://" + in.Host + rawPath + if q := query.Encode(); q != "" { + back += "?" + q + } + login := fmt.Sprintf("%s/auth/r/%s?return=%s", strings.TrimRight(dash, "/"), m.ID, url.QueryEscape(back)) + if share != "" { + login += "&share=" + url.QueryEscape(share) + } + d.Cookie, d.Result = "", "auth-required" + if (in.Method == http.MethodGet || in.Method == http.MethodHead) && !in.Upgrade { + d.Status, d.Location = http.StatusFound, login + } else { + d.Status = http.StatusUnauthorized + } + return d +} + +// callback trades hammerhead's short-lived login token for this visitor's cookie. +func (b *Blacktip) callback(w http.ResponseWriter, r *http.Request, ms *mapState, secure bool) (user, result string) { + b.mu.Lock() + key := b.authKey + b.mu.Unlock() + t, ok := authtoken.Verify(key, r.URL.Query().Get("t"), authtoken.KindCallback, ms.m.ID, ms.epoch.Load()) + if !ok { + http.Error(w, "That login link is invalid or has expired. Go back and sign in again.", http.StatusForbidden) + return "", "bad-login-token" + } + hours := ms.m.Auth.SessionHours + if hours <= 0 { + hours = 24 + } + sess := authtoken.Sign(key, authtoken.Token{Mapping: ms.m.ID, Sub: t.Sub, Kind: authtoken.KindSession, Epoch: t.Epoch, + Exp: time.Now().Add(time.Duration(hours) * time.Hour).Unix()}) + http.SetCookie(w, &http.Cookie{Name: authCookie, Value: sess, Path: "/", HttpOnly: true, Secure: secure, + SameSite: http.SameSiteLaxMode, MaxAge: hours * 3600}) + http.Redirect(w, r, safeDest(r.URL.Query().Get("r")), http.StatusFound) + return t.Sub, "login" +} + +func (b *Blacktip) logout(w http.ResponseWriter, r *http.Request, secure bool) { + http.SetCookie(w, &http.Cookie{Name: authCookie, Value: "", Path: "/", MaxAge: -1, HttpOnly: true, Secure: secure}) + http.Redirect(w, r, "/", http.StatusFound) +} + +// ServeHTTP handles every request on a built-in http/https listener. +func (l *listener) ServeHTTP(w http.ResponseWriter, r *http.Request) { + b := l.b + start := time.Now() + rec := &recorder{ResponseWriter: w} + host := hostOnly(r.Host) + ms, found := route(l.snapshot(), host) + if !found { + http.Error(rec, "Nothing is served at this address.", http.StatusNotFound) + return + } + entry := model.LogEntry{Time: start, Mapping: ms.m.ID, Remote: r.RemoteAddr, Method: r.Method, Host: host, Path: r.URL.Path, Result: "allowed"} + defer func() { + entry.Status = rec.status + if entry.Status == 0 { + entry.Status = 200 + if r.Header.Get("Upgrade") != "" { + entry.Status = 101 + } + } + if r.ContentLength > 0 { + entry.BytesIn = r.ContentLength + } + entry.BytesOut = rec.bytes + entry.Millis = time.Since(start).Milliseconds() + ms.conns.Add(1) + ms.bytesIn.Add(entry.BytesIn) + ms.bytesOut.Add(entry.BytesOut) + b.logAccess(entry) + }() + + secure := l.kind == model.ProtoHTTPS + scheme := "http" + if secure { + scheme = "https" + } + if ms.m.Auth.Protected() { + switch r.URL.Path { + case "/.hammerhead/callback": + entry.User, entry.Result = b.callback(rec, r, ms, secure) + return + case "/.hammerhead/logout": + b.logout(rec, r, secure) + entry.Result = "logout" + return + } + } + + d := b.decide(ms, authInput{Method: r.Method, Scheme: scheme, Host: r.Host, URI: r.URL.RequestURI(), + ClientIP: hostFromAddr(r.RemoteAddr), Cookie: r.Header.Get("Cookie"), Upgrade: r.Header.Get("Upgrade") != ""}) + entry.Result, entry.User = d.Result, d.User + if !d.Allow { + switch d.Status { + case http.StatusFound: + http.Redirect(rec, r, d.Location, http.StatusFound) + case http.StatusForbidden: + http.Error(rec, "Forbidden", http.StatusForbidden) + default: + http.Error(rec, "Authentication required.", http.StatusUnauthorized) + } + return + } + + // the backend never sees our cookie, and can trust X-Hammerhead-User + r.Header.Del("X-Hammerhead-User") + if d.User != "" { + r.Header.Set("X-Hammerhead-User", d.User) + } + if d.Cookie != "" { + r.Header.Set("Cookie", d.Cookie) + } else { + r.Header.Del("Cookie") + } + ms.proxy.ServeHTTP(rec, r) +}