Add the Bamboo site agent
The site agent: a WireGuard peer that dials out to each edge, accepts TCP streams and native UDP datagrams inside the tunnel, lists local Docker containers, and runs the local SMTP relay for mail egress. Split out of the original single repository; depends on Reef for the shared types. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
VeillaxandClaude Sonnet 5.5 committed
1 parent
c93a1e4278
commit
2e7cb8171f
8 files changed
+976
-1
No files matched your search
@@ -0,0 +1,416 @@
|
||||
// Package bamboo is the site agent. It only makes outbound connections: one
|
||||
// HTTP long-poll to hammerhead for instructions, and one WireGuard tunnel
|
||||
// (UDP) to each blacktip edge that forwards ports to this site. Inside the
|
||||
// tunnel it accepts proxied connections and dials the real services.
|
||||
package bamboo
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"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
|
||||
Token string
|
||||
DataDir string // holds this agent's WireGuard key
|
||||
Docker bool // report local Docker containers so they can be picked as targets
|
||||
}
|
||||
|
||||
type Bamboo struct {
|
||||
cfg Config
|
||||
http *http.Client
|
||||
ctx context.Context
|
||||
wgKey string
|
||||
wgPub string
|
||||
|
||||
mu sync.Mutex
|
||||
version int64
|
||||
tunIP string
|
||||
tun *wg.Device
|
||||
edges []model.EdgeRef
|
||||
lastRep string
|
||||
|
||||
mail map[string]*mailRun
|
||||
mailErr map[string]string // mail egress ID -> "" (listening) or an error
|
||||
}
|
||||
|
||||
func New(cfg Config) (*Bamboo, error) {
|
||||
if cfg.DataDir == "" {
|
||||
cfg.DataDir = "./bamboo-data"
|
||||
}
|
||||
key, err := loadOrCreateKey(filepath.Join(cfg.DataDir, "wg.key"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pub, err := wg.PublicKey(key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Bamboo{cfg: cfg, http: &http.Client{Timeout: 40 * time.Second}, ctx: context.Background(), wgKey: key, wgPub: pub,
|
||||
mail: map[string]*mailRun{}, mailErr: map[string]string{}}, nil
|
||||
}
|
||||
|
||||
// loadOrCreateKey keeps this agent'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)
|
||||
}
|
||||
|
||||
func (b *Bamboo) Run(ctx context.Context) error {
|
||||
b.ctx = ctx
|
||||
log.Printf("bamboo started; WireGuard public key %s", b.wgPub)
|
||||
go b.reportLoop(ctx)
|
||||
go b.refreshLoop(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 {
|
||||
b.apply(cfg)
|
||||
}
|
||||
}
|
||||
b.mu.Lock()
|
||||
for _, r := range b.mail {
|
||||
r.ln.Close()
|
||||
}
|
||||
if b.tun != nil {
|
||||
b.tun.Close()
|
||||
}
|
||||
b.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *Bamboo) fetchConfig(ctx context.Context) (*model.SiteConfig, error) {
|
||||
b.mu.Lock()
|
||||
v := b.version
|
||||
b.mu.Unlock()
|
||||
req, _ := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/site/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.SiteConfig
|
||||
return &cfg, json.NewDecoder(resp.Body).Decode(&cfg)
|
||||
}
|
||||
|
||||
func (b *Bamboo) apply(cfg *model.SiteConfig) {
|
||||
b.mu.Lock()
|
||||
b.version = cfg.Version
|
||||
b.edges = cfg.Edges
|
||||
if cfg.TunnelIP != b.tunIP && b.tun != nil {
|
||||
b.tun.Close()
|
||||
b.tun = nil
|
||||
}
|
||||
if b.tun == nil && cfg.TunnelIP != "" {
|
||||
if err := b.startTunnel(cfg.TunnelIP); err != nil {
|
||||
log.Printf("start WireGuard: %v", err)
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
b.syncPeers()
|
||||
b.syncMail(cfg.MailEgress)
|
||||
go b.report()
|
||||
}
|
||||
|
||||
// startTunnel creates the WireGuard device and starts serving inside it. Caller holds b.mu.
|
||||
func (b *Bamboo) startTunnel(ip string) error {
|
||||
addr, err := netip.ParseAddr(ip)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dev, err := wg.New(b.wgKey, addr, 0) // we dial out, so any local UDP port will do
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tcp, err := dev.ListenTCP(tunnel.ProxyTCPPort)
|
||||
if err != nil {
|
||||
dev.Close()
|
||||
return err
|
||||
}
|
||||
udp, err := dev.ListenUDP(tunnel.ProxyUDPPort)
|
||||
if err != nil {
|
||||
dev.Close()
|
||||
return err
|
||||
}
|
||||
b.tun, b.tunIP = dev, ip
|
||||
go serveTCP(tcp)
|
||||
go serveUDP(udp)
|
||||
log.Printf("WireGuard up at %s", ip)
|
||||
return nil
|
||||
}
|
||||
|
||||
// syncPeers makes the tunnel match the edges hammerhead lists. Unchanged peers keep their sessions.
|
||||
func (b *Bamboo) syncPeers() {
|
||||
b.mu.Lock()
|
||||
tun, edges := b.tun, append([]model.EdgeRef(nil), b.edges...)
|
||||
b.mu.Unlock()
|
||||
if tun == nil {
|
||||
return
|
||||
}
|
||||
var peers []wg.Peer
|
||||
for _, e := range edges {
|
||||
if e.PublicKey == "" || e.IP == "" {
|
||||
continue // that edge hasn't started yet
|
||||
}
|
||||
peers = append(peers, wg.Peer{PublicKey: e.PublicKey, AllowedIP: e.IP, Endpoint: e.Endpoint})
|
||||
}
|
||||
if err := tun.SetPeers(peers); err != nil {
|
||||
log.Printf("peers: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// refreshLoop re-resolves edge hostnames, so a changed DNS record is followed.
|
||||
func (b *Bamboo) refreshLoop(ctx context.Context) {
|
||||
t := time.NewTicker(60 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
b.syncPeers()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// connected lists the edges with a recent WireGuard handshake.
|
||||
func (b *Bamboo) connected() []string {
|
||||
b.mu.Lock()
|
||||
tun, edges := b.tun, b.edges
|
||||
b.mu.Unlock()
|
||||
ids := []string{}
|
||||
if tun == nil {
|
||||
return ids
|
||||
}
|
||||
hs := tun.Handshakes()
|
||||
for _, e := range edges {
|
||||
if t, ok := hs[e.PublicKey]; ok && time.Since(t) < 3*time.Minute {
|
||||
ids = append(ids, e.ID)
|
||||
}
|
||||
}
|
||||
sort.Strings(ids)
|
||||
return ids
|
||||
}
|
||||
|
||||
func (b *Bamboo) report() {
|
||||
ids := b.connected()
|
||||
var containers []model.Container
|
||||
if b.cfg.Docker {
|
||||
ctx, cancel := context.WithTimeout(b.ctx, 5*time.Second)
|
||||
containers, _ = docker.List(ctx)
|
||||
cancel()
|
||||
}
|
||||
body, _ := json.Marshal(model.SiteReport{PublicKey: b.wgPub, Connected: ids, Containers: containers, MailEgress: b.mailStatus()})
|
||||
req, _ := http.NewRequest("POST", b.cfg.Hammerhead+"/site/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 {
|
||||
return
|
||||
}
|
||||
resp.Body.Close()
|
||||
}
|
||||
|
||||
// reportLoop reports every 20 seconds, and straight away when the set of connected edges changes.
|
||||
func (b *Bamboo) reportLoop(ctx context.Context) {
|
||||
b.report()
|
||||
slow := time.NewTicker(20 * time.Second)
|
||||
fast := time.NewTicker(3 * time.Second)
|
||||
defer slow.Stop()
|
||||
defer fast.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-slow.C:
|
||||
b.report()
|
||||
case <-fast.C:
|
||||
sig := strings.Join(b.connected(), ",")
|
||||
b.mu.Lock()
|
||||
changed := sig != b.lastRep
|
||||
b.lastRep = sig
|
||||
b.mu.Unlock()
|
||||
if changed {
|
||||
b.report()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- serving proxied connections ----------
|
||||
|
||||
func serveTCP(ln net.Listener) {
|
||||
for {
|
||||
c, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
go handleStream(c)
|
||||
}
|
||||
}
|
||||
|
||||
func handleStream(st net.Conn) {
|
||||
st.SetReadDeadline(time.Now().Add(10 * time.Second))
|
||||
br := bufio.NewReader(st)
|
||||
h, err := tunnel.ReadHeader(br)
|
||||
st.SetReadDeadline(time.Time{})
|
||||
if err != nil || h.Proto != "tcp" {
|
||||
st.Close()
|
||||
return
|
||||
}
|
||||
c, err := net.DialTimeout("tcp", net.JoinHostPort(h.Host, strconv.Itoa(h.Port)), 8*time.Second)
|
||||
if err != nil {
|
||||
st.Close() // the edge reads this as "target unreachable"
|
||||
return
|
||||
}
|
||||
tunnel.Proxy(st, br, c, c)
|
||||
}
|
||||
|
||||
type udpSession struct {
|
||||
conn net.Conn
|
||||
last time.Time
|
||||
}
|
||||
|
||||
// serveUDP relays datagrams between the tunnel and UDP targets. Each datagram
|
||||
// starts with a type byte: 0 = target header + payload (creates the session),
|
||||
// 1 = payload for an existing session. Sessions are keyed by the edge's source port.
|
||||
func serveUDP(pc net.PacketConn) {
|
||||
var mu sync.Mutex
|
||||
sessions := map[string]*udpSession{}
|
||||
go func() {
|
||||
t := time.NewTicker(15 * time.Second)
|
||||
defer t.Stop()
|
||||
for range t.C {
|
||||
mu.Lock()
|
||||
for _, s := range sessions {
|
||||
if time.Since(s.last) > 60*time.Second {
|
||||
s.conn.Close()
|
||||
}
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
buf := make([]byte, 65535)
|
||||
for {
|
||||
n, from, err := pc.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if n < 2 {
|
||||
continue
|
||||
}
|
||||
key := from.String()
|
||||
data := buf[:n]
|
||||
var payload []byte
|
||||
mu.Lock()
|
||||
s := sessions[key]
|
||||
switch data[0] {
|
||||
case 0:
|
||||
if len(data) < 3 {
|
||||
mu.Unlock()
|
||||
continue
|
||||
}
|
||||
hl := int(data[1])<<8 | int(data[2])
|
||||
if len(data) < 3+hl {
|
||||
mu.Unlock()
|
||||
continue
|
||||
}
|
||||
payload = data[3+hl:]
|
||||
if s == nil {
|
||||
var h tunnel.Header
|
||||
if json.Unmarshal(data[3:3+hl], &h) != nil || h.Proto != "udp" {
|
||||
mu.Unlock()
|
||||
continue
|
||||
}
|
||||
c, err := net.Dial("udp", net.JoinHostPort(h.Host, strconv.Itoa(h.Port)))
|
||||
if err != nil {
|
||||
mu.Unlock()
|
||||
continue
|
||||
}
|
||||
s = &udpSession{conn: c}
|
||||
sessions[key] = s
|
||||
go func(s *udpSession, from net.Addr, key string) {
|
||||
defer func() {
|
||||
s.conn.Close()
|
||||
mu.Lock()
|
||||
if sessions[key] == s {
|
||||
delete(sessions, key)
|
||||
}
|
||||
mu.Unlock()
|
||||
}()
|
||||
rb := make([]byte, 65535)
|
||||
for {
|
||||
n, err := s.conn.Read(rb)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
s.last = time.Now()
|
||||
mu.Unlock()
|
||||
if _, err := pc.WriteTo(rb[:n], from); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}(s, from, key)
|
||||
}
|
||||
case 1:
|
||||
payload = data[1:]
|
||||
default:
|
||||
mu.Unlock()
|
||||
continue
|
||||
}
|
||||
if s != nil {
|
||||
s.last = time.Now()
|
||||
}
|
||||
mu.Unlock()
|
||||
if s != nil && len(payload) > 0 {
|
||||
s.conn.Write(payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user