Add Spinner, a Minecraft hostname-routing reverse proxy
CI / test (push) Successful in 2m32s

Routes Java Edition connections by the hostname in the handshake, with
wildcard domains, backend failover, PROXY protocol v2, cached server-list
status from health probes, hot-reloaded config, and an admin page and
metrics endpoint. Includes CI, tagged releases and a container image.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
VeillaxandClaude Sonnet 5.5 committed 2026-10-26 19:40:40 -06:00
1 parent e640c5bf02
commit b53af0e196
29 files changed
+2640

No files matched your search

+140
View File
@@ -0,0 +1,140 @@
// Package admin serves the REST API, Prometheus metrics and the embedded UI.
package admin
import (
"crypto/sha256"
"crypto/subtle"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"git.veillax.com/Veillax/spinner/internal/config"
"git.veillax.com/Veillax/spinner/internal/health"
"git.veillax.com/Veillax/spinner/internal/proxy"
"git.veillax.com/Veillax/spinner/web"
)
type Admin struct {
Cfg *config.Manager
Health *health.Tracker
Proxy *proxy.Proxy
Token string
}
func (a *Admin) Handler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Header().Set("Content-Security-Policy", "default-src 'self'; style-src 'unsafe-inline'; script-src 'unsafe-inline'")
w.Write(web.Index)
})
mux.HandleFunc("GET /api/overview", a.auth(a.overview))
mux.HandleFunc("GET /api/config", a.auth(a.getConfig))
mux.HandleFunc("PUT /api/config", a.auth(a.putConfig))
mux.HandleFunc("GET /metrics", a.auth(a.metrics))
return mux
}
func (a *Admin) auth(next http.HandlerFunc) http.HandlerFunc {
want := sha256.Sum256([]byte(a.Token))
return func(w http.ResponseWriter, r *http.Request) {
got := sha256.Sum256([]byte(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")))
if subtle.ConstantTimeCompare(got[:], want[:]) != 1 {
w.Header().Set("WWW-Authenticate", `Bearer realm="spinner"`)
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
next(w, r)
}
}
func writeJSON(w http.ResponseWriter, v any) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(v)
}
type serverView struct {
config.Server
Health *health.State `json:"health,omitempty"`
Stats *proxy.ServerSnapshot `json:"stats,omitempty"`
}
func (a *Admin) overview(w http.ResponseWriter, r *http.Request) {
snap := a.Proxy.Stats.Snapshot()
states := map[string]health.State{}
for _, s := range a.Health.Snapshot() {
states[s.Name] = s
}
stats := map[string]proxy.ServerSnapshot{}
for _, s := range snap.Servers {
stats[s.Name] = s
}
var servers []serverView
for _, s := range a.Cfg.Current().Servers {
v := serverView{Server: s}
if h, ok := states[s.Name]; ok {
v.Health = &h
}
if st, ok := stats[s.Name]; ok {
v.Stats = &st
}
servers = append(servers, v)
}
snap.Servers = nil
writeJSON(w, map[string]any{"totals": snap, "servers": servers})
}
func (a *Admin) getConfig(w http.ResponseWriter, r *http.Request) { writeJSON(w, a.Cfg.Current()) }
func (a *Admin) putConfig(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
cfg, err := config.Parse(body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := a.Cfg.Save(cfg); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
writeJSON(w, cfg)
}
func esc(s string) string {
return strings.NewReplacer(`\`, `\\`, `"`, `\"`, "\n", `\n`).Replace(s)
}
func (a *Admin) metrics(w http.ResponseWriter, r *http.Request) {
snap := a.Proxy.Stats.Snapshot()
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
p := func(format string, args ...any) { fmt.Fprintf(w, format+"\n", args...) }
p("# TYPE spinner_uptime_seconds gauge\nspinner_uptime_seconds %d", snap.UptimeSecs)
p("# TYPE spinner_connections_active gauge\nspinner_connections_active %d", snap.Active)
p("# TYPE spinner_connections_total counter\nspinner_connections_total %d", snap.Total)
p("# TYPE spinner_connections_rejected_total counter\nspinner_connections_rejected_total %d", snap.Rejected)
p("# TYPE spinner_unknown_host_total counter\nspinner_unknown_host_total %d", snap.Unknown)
for _, s := range snap.Servers {
l := fmt.Sprintf(`{server="%s"}`, esc(s.Name))
p("spinner_server_connections_active%s %d", l, s.Active)
p("spinner_server_connections_total%s %d", l, s.Total)
p("spinner_server_failures_total%s %d", l, s.Failures)
p("spinner_server_bytes_in_total%s %d", l, s.BytesIn)
p("spinner_server_bytes_out_total%s %d", l, s.BytesOut)
}
for _, h := range a.Health.Snapshot() {
up := 0
if h.Up {
up = 1
}
l := fmt.Sprintf(`{server="%s"}`, esc(h.Name))
p("spinner_server_up%s %d", l, up)
p("spinner_server_latency_ms%s %d", l, h.LatencyMS)
p("spinner_server_players_online%s %d", l, h.Online)
}
}
+219
View File
@@ -0,0 +1,219 @@
// Package config loads, validates, saves and hot-reloads Spinner's JSON config.
package config
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"time"
)
type Config struct {
Bind string `json:"bind"`
AdminBind string `json:"admin_bind"`
HandshakeTimeoutSecs int `json:"handshake_timeout_secs"`
ConnectTimeoutSecs int `json:"connect_timeout_secs"`
ProbeIntervalSecs int `json:"probe_interval_secs"`
MaxConnsPerIP int `json:"max_conns_per_ip"`
UnknownMOTD string `json:"unknown_motd"`
Servers []Server `json:"servers"`
}
type Server struct {
Name string `json:"name"`
Domains []string `json:"domains"`
Addresses []string `json:"addresses"`
SendProxyProtocol bool `json:"send_proxy_protocol,omitempty"`
OfflineMOTD string `json:"offline_motd,omitempty"`
MaxConnections int `json:"max_connections,omitempty"`
}
func (c *Config) HandshakeTimeout() time.Duration {
return time.Duration(c.HandshakeTimeoutSecs) * time.Second
}
func (c *Config) ConnectTimeout() time.Duration {
return time.Duration(c.ConnectTimeoutSecs) * time.Second
}
func (c *Config) ProbeInterval() time.Duration {
return time.Duration(c.ProbeIntervalSecs) * time.Second
}
// Parse decodes, applies defaults to and validates a config document.
func Parse(data []byte) (*Config, error) {
var c Config
dec := json.NewDecoder(bytes.NewReader(data))
dec.DisallowUnknownFields()
if err := dec.Decode(&c); err != nil {
return nil, err
}
def := func(v *int, d int) {
if *v <= 0 {
*v = d
}
}
if c.Bind == "" {
c.Bind = ":25565"
}
def(&c.HandshakeTimeoutSecs, 5)
def(&c.ConnectTimeoutSecs, 3)
def(&c.ProbeIntervalSecs, 10)
if c.UnknownMOTD == "" {
c.UnknownMOTD = "§cUnknown server"
}
return &c, c.validate()
}
func (c *Config) validate() error {
seen := map[string]bool{}
for i := range c.Servers {
s := &c.Servers[i]
if s.Name == "" {
return fmt.Errorf("server %d has no name. Give every server a name", i+1)
}
if seen[s.Name] {
return fmt.Errorf("two servers are named %q. Names must be different", s.Name)
}
seen[s.Name] = true
if len(s.Domains) == 0 {
return fmt.Errorf("server %q has no addresses for players. Add at least one", s.Name)
}
if len(s.Addresses) == 0 {
return fmt.Errorf("server %q has no backend address. Add at least one as host:port", s.Name)
}
for _, a := range s.Addresses {
if _, _, err := net.SplitHostPort(a); err != nil {
return fmt.Errorf("server %q: backend address %q is not host:port (%v)", s.Name, a, err)
}
}
for _, d := range s.Domains {
if strings.TrimSpace(d) == "" || strings.Contains(strings.TrimPrefix(d, "*."), "*") {
return fmt.Errorf("server %q: %q is not a valid address for players. Use a name like play.example.com or *.example.com", s.Name, d)
}
}
}
return nil
}
// Manager owns the current config, persists changes and reloads on file edits.
type Manager struct {
path string
cur atomic.Pointer[Config]
mu sync.Mutex // guards raw and subs
raw []byte
subs []func(*Config)
log *slog.Logger
}
func NewManager(path string, log *slog.Logger) (*Manager, error) {
data, err := os.ReadFile(path)
if err != nil {
return nil, err
}
c, err := Parse(data)
if err != nil {
return nil, fmt.Errorf("%s: %w", path, err)
}
m := &Manager{path: path, raw: data, log: log}
m.cur.Store(c)
return m, nil
}
func (m *Manager) Current() *Config { return m.cur.Load() }
// OnChange registers a callback invoked after every successful reload or save.
func (m *Manager) OnChange(f func(*Config)) {
m.mu.Lock()
m.subs = append(m.subs, f)
m.mu.Unlock()
}
func (m *Manager) apply(data []byte, c *Config) {
m.mu.Lock()
m.raw = data
subs := append([]func(*Config){}, m.subs...)
m.mu.Unlock()
m.cur.Store(c)
for _, f := range subs {
f(c)
}
}
// Save validates c, writes it atomically and applies it immediately.
func (m *Manager) Save(c *Config) error {
data, err := json.MarshalIndent(c, "", " ")
if err != nil {
return err
}
parsed, err := Parse(data)
if err != nil {
return err
}
tmp, err := os.CreateTemp(filepath.Dir(m.path), ".spinner-*.json")
if err != nil {
return err
}
_, werr := tmp.Write(append(data, '\n'))
cerr := tmp.Close()
if err := errors.Join(werr, cerr); err != nil {
os.Remove(tmp.Name())
return err
}
if err := os.Rename(tmp.Name(), m.path); err != nil {
os.Remove(tmp.Name())
return err
}
m.mu.Lock()
m.raw = append(data, '\n')
m.mu.Unlock()
m.apply(m.raw, parsed)
return nil
}
func (m *Manager) reloadIfChanged() {
data, err := os.ReadFile(m.path)
if err != nil {
return
}
m.mu.Lock()
same := bytes.Equal(data, m.raw)
m.mu.Unlock()
if same {
return
}
c, err := Parse(data)
if err != nil {
m.log.Error("config reload rejected, keeping previous config", "err", err)
m.mu.Lock()
m.raw = data // don't repeat the error until the file changes again
m.mu.Unlock()
return
}
if old := m.Current(); old.Bind != c.Bind || old.AdminBind != c.AdminBind {
m.log.Warn("bind address changes need a restart")
}
m.apply(data, c)
m.log.Info("config reloaded", "servers", len(c.Servers))
}
// Watch polls the file until ctx ends.
func (m *Manager) Watch(ctx context.Context) {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
m.reloadIfChanged()
}
}
}
+154
View File
@@ -0,0 +1,154 @@
// Package health probes backends with a status ping, caching their server-list
// response and tracking whether each one is up.
package health
import (
"context"
"encoding/json"
"log/slog"
"net"
"sort"
"sync"
"time"
"git.veillax.com/Veillax/spinner/internal/config"
"git.veillax.com/Veillax/spinner/internal/protocol"
)
type State struct {
Name string `json:"name"`
Up bool `json:"up"`
LastCheck time.Time `json:"last_check"`
LatencyMS int64 `json:"latency_ms"`
Error string `json:"error,omitempty"`
Version string `json:"version,omitempty"`
Online int `json:"online"`
Max int `json:"max"`
status string
}
type Tracker struct {
mu sync.RWMutex
st map[string]*State
log *slog.Logger
}
func New(log *slog.Logger) *Tracker {
return &Tracker{st: map[string]*State{}, log: log}
}
// Cached returns the backend's last status JSON if it is up and the probe is
// no older than maxAge.
func (t *Tracker) Cached(name string, maxAge time.Duration) (string, bool) {
t.mu.RLock()
defer t.mu.RUnlock()
s := t.st[name]
if s == nil || !s.Up || s.status == "" || time.Since(s.LastCheck) > maxAge {
return "", false
}
return s.status, true
}
func (t *Tracker) Snapshot() []State {
t.mu.RLock()
defer t.mu.RUnlock()
out := make([]State, 0, len(t.st))
for _, s := range t.st {
out = append(out, *s)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out
}
// Run probes every server immediately and then on each interval until ctx ends.
func (t *Tracker) Run(ctx context.Context, m *config.Manager) {
for {
cfg := m.Current()
t.probeAll(ctx, cfg)
select {
case <-ctx.Done():
return
case <-time.After(cfg.ProbeInterval()):
}
}
}
func (t *Tracker) probeAll(ctx context.Context, cfg *config.Config) {
var wg sync.WaitGroup
for i := range cfg.Servers {
wg.Add(1)
go func(s *config.Server) {
defer wg.Done()
t.record(s.Name, Probe(ctx, s, cfg.ConnectTimeout()))
}(&cfg.Servers[i])
}
wg.Wait()
names := map[string]bool{}
for _, s := range cfg.Servers {
names[s.Name] = true
}
t.mu.Lock()
for n := range t.st {
if !names[n] {
delete(t.st, n)
}
}
t.mu.Unlock()
}
// Result is the outcome of one probe.
type Result struct {
Status string
Latency time.Duration
Err error
}
// Probe tries each of the server's addresses in order and returns the first success.
func Probe(ctx context.Context, s *config.Server, timeout time.Duration) Result {
var last Result
for _, addr := range s.Addresses {
start := time.Now()
d := net.Dialer{Timeout: timeout}
conn, err := d.DialContext(ctx, "tcp", addr)
if err != nil {
last = Result{Err: err}
continue
}
host, port := protocol.SplitHostPort(addr)
js, err := protocol.QueryStatus(conn, host, port, timeout)
conn.Close()
if err != nil {
last = Result{Err: err}
continue
}
return Result{Status: js, Latency: time.Since(start)}
}
return last
}
func (t *Tracker) record(name string, r Result) {
s := State{Name: name, Up: r.Err == nil, LastCheck: time.Now(), LatencyMS: r.Latency.Milliseconds(), status: r.Status}
if r.Err != nil {
s.Error = r.Err.Error()
} else {
var parsed struct {
Version struct{ Name string } `json:"version"`
Players struct{ Online, Max int } `json:"players"`
}
if json.Unmarshal([]byte(r.Status), &parsed) == nil {
s.Version, s.Online, s.Max = parsed.Version.Name, parsed.Players.Online, parsed.Players.Max
}
}
t.mu.Lock()
prev := t.st[name]
t.st[name] = &s
t.mu.Unlock()
if prev == nil || prev.Up != s.Up {
if s.Up {
t.log.Info("backend up", "server", name, "latency_ms", s.LatencyMS)
} else {
t.log.Warn("backend down", "server", name, "err", s.Error)
}
}
}
+215
View File
@@ -0,0 +1,215 @@
// Package protocol implements the small slice of the Minecraft Java protocol
// Spinner needs: the handshake, the status exchange and the login disconnect.
// Everything after the handshake is passed through untouched.
package protocol
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"strconv"
"time"
)
const (
maxHandshake = 1024
maxStatus = 1 << 20
)
// ErrLegacy is returned for pre-1.7 server list pings, which Spinner does not route.
var ErrLegacy = errors.New("legacy ping")
type Handshake struct {
Protocol int32
Host string
Port uint16
NextState int32 // 1 status, 2 login, 3 transfer
}
func (h Handshake) IsStatus() bool { return h.NextState == 1 }
func readVarInt(r io.Reader, raw *[]byte) (int32, error) {
var v uint32
var b [1]byte
for i := 0; i < 5; i++ {
if _, err := io.ReadFull(r, b[:]); err != nil {
return 0, err
}
if raw != nil {
*raw = append(*raw, b[0])
}
v |= uint32(b[0]&0x7f) << (7 * i)
if b[0]&0x80 == 0 {
return int32(v), nil
}
}
return 0, errors.New("varint too long")
}
func AppendVarInt(buf []byte, v int32) []byte {
u := uint32(v)
for u >= 0x80 {
buf = append(buf, byte(u)|0x80)
u >>= 7
}
return append(buf, byte(u))
}
func appendString(buf []byte, s string) []byte {
buf = AppendVarInt(buf, int32(len(s)))
return append(buf, s...)
}
// Frame prefixes a packet body (id + payload) with its length.
func Frame(body []byte) []byte {
return append(AppendVarInt(make([]byte, 0, len(body)+5), int32(len(body))), body...)
}
// ReadHandshake reads one handshake packet and returns it along with the exact
// bytes consumed, so they can be replayed to the backend unchanged.
func ReadHandshake(r io.Reader) (Handshake, []byte, error) {
var raw []byte
n, err := readVarInt(r, &raw)
if err != nil {
return Handshake{}, raw, err
}
if len(raw) > 0 && raw[0] == 0xFE {
return Handshake{}, raw, ErrLegacy
}
if n <= 0 || n > maxHandshake {
return Handshake{}, raw, fmt.Errorf("invalid handshake length %d", n)
}
body := make([]byte, n)
if _, err := io.ReadFull(r, body); err != nil {
return Handshake{}, raw, err
}
raw = append(raw, body...)
br := bytes.NewReader(body)
id, err := readVarInt(br, nil)
if err != nil || id != 0 {
return Handshake{}, raw, errors.New("expected handshake packet")
}
var h Handshake
if h.Protocol, err = readVarInt(br, nil); err != nil {
return Handshake{}, raw, err
}
hl, err := readVarInt(br, nil)
if err != nil || hl < 0 || int(hl) > br.Len()-3 {
return Handshake{}, raw, errors.New("invalid host length")
}
host := make([]byte, hl)
io.ReadFull(br, host)
h.Host = string(host)
var port [2]byte
io.ReadFull(br, port[:])
h.Port = uint16(port[0])<<8 | uint16(port[1])
if h.NextState, err = readVarInt(br, nil); err != nil {
return Handshake{}, raw, err
}
return h, raw, nil
}
func BuildHandshake(protocol int32, host string, port uint16, next int32) []byte {
body := AppendVarInt(nil, 0)
body = AppendVarInt(body, protocol)
body = appendString(body, host)
body = append(body, byte(port>>8), byte(port))
body = AppendVarInt(body, next)
return Frame(body)
}
// ReadPacket reads one length-prefixed packet and returns its id and payload.
func ReadPacket(r io.Reader, max int32) (int32, []byte, error) {
n, err := readVarInt(r, nil)
if err != nil {
return 0, nil, err
}
if n <= 0 || n > max {
return 0, nil, fmt.Errorf("invalid packet length %d", n)
}
body := make([]byte, n)
if _, err := io.ReadFull(r, body); err != nil {
return 0, nil, err
}
br := bytes.NewReader(body)
id, err := readVarInt(br, nil)
if err != nil {
return 0, nil, err
}
return id, body[len(body)-br.Len():], nil
}
func stringPacket(id int32, s string) []byte {
return Frame(appendString(AppendVarInt(nil, id), s))
}
// StatusJSON builds a server list response.
func StatusJSON(clientProtocol int32, version, motd string, online, max int) string {
b, _ := json.Marshal(map[string]any{
"version": map[string]any{"name": version, "protocol": clientProtocol},
"players": map[string]any{"max": max, "online": online},
"description": map[string]any{"text": motd},
})
return string(b)
}
// LoginDisconnect is the packet a client shows as "Disconnected: <message>".
func LoginDisconnect(message string) []byte {
b, _ := json.Marshal(map[string]string{"text": message})
return stringPacket(0, string(b))
}
// ServeStatus plays the server side of the status exchange with a canned
// response: request, response, then an echoed ping if the client sends one.
func ServeStatus(rw io.ReadWriter, statusJSON string) error {
if id, _, err := ReadPacket(rw, 16); err != nil || id != 0 {
return errors.New("bad status request")
}
if _, err := rw.Write(stringPacket(0, statusJSON)); err != nil {
return err
}
id, payload, err := ReadPacket(rw, 32)
if err != nil || id != 1 {
return nil // client closed after the response
}
_, err = rw.Write(Frame(append(AppendVarInt(nil, 1), payload...)))
return err
}
// QueryStatus performs a status ping against a backend and returns the raw JSON.
func QueryStatus(conn net.Conn, host string, port uint16, timeout time.Duration) (string, error) {
conn.SetDeadline(time.Now().Add(timeout))
if _, err := conn.Write(BuildHandshake(767, host, port, 1)); err != nil {
return "", err
}
if _, err := conn.Write(Frame([]byte{0})); err != nil {
return "", err
}
id, payload, err := ReadPacket(conn, maxStatus)
if err != nil {
return "", err
}
if id != 0 {
return "", fmt.Errorf("unexpected packet id %d", id)
}
br := bytes.NewReader(payload)
n, err := readVarInt(br, nil)
if err != nil || n < 0 || int(n) > br.Len() {
return "", errors.New("bad status string")
}
return string(payload[len(payload)-br.Len() : len(payload)-br.Len()+int(n)]), nil
}
// SplitHostPort parses a backend "host:port" address.
func SplitHostPort(addr string) (string, uint16) {
h, p, err := net.SplitHostPort(addr)
if err != nil {
return addr, 25565
}
n, _ := strconv.Atoi(p)
return h, uint16(n)
}
+87
View File
@@ -0,0 +1,87 @@
package protocol
import (
"bytes"
"net"
"testing"
"time"
)
func TestHandshakeRoundTrip(t *testing.T) {
in := BuildHandshake(767, "mc.example.com", 25565, 2)
h, raw, err := ReadHandshake(bytes.NewReader(in))
if err != nil {
t.Fatal(err)
}
want := Handshake{Protocol: 767, Host: "mc.example.com", Port: 25565, NextState: 2}
if h != want {
t.Fatalf("got %+v want %+v", h, want)
}
if !bytes.Equal(raw, in) {
t.Fatal("raw bytes not preserved")
}
}
func TestReadHandshakeRejects(t *testing.T) {
cases := map[string][]byte{
"legacy": {0xfe, 0x01, 0xfa},
"huge length": {0xff, 0xff, 0xff, 0xff, 0x07},
"zero length": {0x00},
"wrong id": Frame([]byte{0x05, 0x00}),
"short host": Frame([]byte{0x00, 0x01, 0x40, 'a'}),
"truncated": {0x10, 0x00},
}
for name, b := range cases {
if _, _, err := ReadHandshake(bytes.NewReader(b)); err == nil {
t.Errorf("%s: expected error", name)
}
}
if _, _, err := ReadHandshake(bytes.NewReader([]byte{0xfe, 0x01})); err != ErrLegacy {
t.Errorf("want ErrLegacy, got %v", err)
}
}
func TestVarInt(t *testing.T) {
if got := AppendVarInt(nil, 300); !bytes.Equal(got, []byte{0xac, 0x02}) {
t.Fatalf("got %x", got)
}
if got := AppendVarInt(nil, -1); len(got) != 5 {
t.Fatalf("negative should take 5 bytes, got %x", got)
}
}
func TestStatusExchange(t *testing.T) {
c, s := net.Pipe()
defer c.Close()
go func() {
defer s.Close()
ServeStatus(s, `{"ok":true}`)
}()
c.SetDeadline(time.Now().Add(2 * time.Second))
c.Write(Frame([]byte{0}))
id, payload, err := ReadPacket(c, maxStatus)
if err != nil || id != 0 || !bytes.Contains(payload, []byte(`{"ok":true}`)) {
t.Fatalf("id=%d payload=%q err=%v", id, payload, err)
}
c.Write(Frame(append([]byte{1}, 0, 0, 0, 0, 0, 0, 0, 42)))
id, payload, err = ReadPacket(c, 32)
if err != nil || id != 1 || payload[7] != 42 {
t.Fatalf("pong id=%d payload=%x err=%v", id, payload, err)
}
}
func TestQueryStatus(t *testing.T) {
c, s := net.Pipe()
defer c.Close()
go func() {
defer s.Close()
if _, _, err := ReadHandshake(s); err != nil {
return
}
ServeStatus(s, `{"players":{"online":3}}`)
}()
got, err := QueryStatus(c, "x", 25565, 2*time.Second)
if err != nil || got != `{"players":{"online":3}}` {
t.Fatalf("got %q err=%v", got, err)
}
}
+190
View File
@@ -0,0 +1,190 @@
// Package proxy accepts Minecraft connections, routes them by hostname and
// pipes bytes to the chosen backend.
package proxy
import (
"context"
"errors"
"io"
"log/slog"
"net"
"sync"
"sync/atomic"
"time"
"git.veillax.com/Veillax/spinner/internal/config"
"git.veillax.com/Veillax/spinner/internal/health"
"git.veillax.com/Veillax/spinner/internal/protocol"
"git.veillax.com/Veillax/spinner/internal/router"
)
type Proxy struct {
cfg *config.Manager
health *health.Tracker
log *slog.Logger
rt atomic.Pointer[router.Router]
Stats *Stats
ipMu sync.Mutex
ipConns map[string]int
}
func New(cfg *config.Manager, h *health.Tracker, log *slog.Logger) *Proxy {
p := &Proxy{cfg: cfg, health: h, log: log, Stats: newStats(), ipConns: map[string]int{}}
p.rt.Store(router.New(cfg.Current()))
cfg.OnChange(func(c *config.Config) { p.rt.Store(router.New(c)) })
return p
}
// Serve accepts connections until ctx ends or the listener fails.
func (p *Proxy) Serve(ctx context.Context, ln net.Listener) error {
go func() { <-ctx.Done(); ln.Close() }()
for {
c, err := ln.Accept()
if err != nil {
if ctx.Err() != nil {
return nil
}
var ne net.Error
if errors.As(err, &ne) && ne.Timeout() {
continue
}
return err
}
go p.handle(c)
}
}
func (p *Proxy) acquireIP(ip string, limit int) bool {
p.ipMu.Lock()
defer p.ipMu.Unlock()
if limit > 0 && p.ipConns[ip] >= limit {
return false
}
p.ipConns[ip]++
return true
}
func (p *Proxy) releaseIP(ip string) {
p.ipMu.Lock()
defer p.ipMu.Unlock()
if p.ipConns[ip]--; p.ipConns[ip] <= 0 {
delete(p.ipConns, ip)
}
}
// reject answers a status ping with a canned MOTD, or a login with a disconnect message.
func (p *Proxy) reject(c net.Conn, h protocol.Handshake, msg string) {
c.SetDeadline(time.Now().Add(10 * time.Second))
if h.IsStatus() {
protocol.ServeStatus(c, protocol.StatusJSON(h.Protocol, "Spinner", msg, 0, 0))
return
}
c.Write(protocol.LoginDisconnect(msg))
}
func (p *Proxy) handle(c net.Conn) {
defer c.Close()
cfg := p.cfg.Current()
p.Stats.total.Add(1)
ip, _, _ := net.SplitHostPort(c.RemoteAddr().String())
if !p.acquireIP(ip, cfg.MaxConnsPerIP) {
p.Stats.rejected.Add(1)
return
}
defer p.releaseIP(ip)
c.SetReadDeadline(time.Now().Add(cfg.HandshakeTimeout()))
hs, raw, err := protocol.ReadHandshake(c)
if err != nil {
p.log.Debug("bad handshake", "remote", ip, "err", err)
return
}
c.SetReadDeadline(time.Time{})
srv := p.rt.Load().Match(hs.Host)
if srv == nil {
p.Stats.unknown.Add(1)
p.log.Debug("unknown host", "remote", ip, "host", hs.Host)
p.reject(c, hs, cfg.UnknownMOTD)
return
}
ss := p.Stats.server(srv.Name)
ss.total.Add(1)
if srv.MaxConnections > 0 && ss.active.Load() >= int64(srv.MaxConnections) {
p.Stats.rejected.Add(1)
p.reject(c, hs, "§cServer is full")
return
}
if hs.IsStatus() {
if js, ok := p.health.Cached(srv.Name, 3*cfg.ProbeInterval()); ok {
c.SetDeadline(time.Now().Add(10 * time.Second))
protocol.ServeStatus(c, js)
return
}
}
b, err := dialAny(srv.Addresses, cfg.ConnectTimeout())
if err != nil {
ss.failures.Add(1)
p.log.Warn("backend unreachable", "server", srv.Name, "err", err)
msg := srv.OfflineMOTD
if msg == "" {
msg = "§c" + srv.Name + " is offline"
}
p.reject(c, hs, msg)
return
}
defer b.Close()
if srv.SendProxyProtocol {
if _, err := b.Write(proxyV2Header(c.RemoteAddr(), c.LocalAddr())); err != nil {
return
}
}
if _, err := b.Write(raw); err != nil {
return
}
p.Stats.active.Add(1)
ss.active.Add(1)
defer func() { p.Stats.active.Add(-1); ss.active.Add(-1) }()
if !hs.IsStatus() {
p.log.Info("connected", "remote", ip, "host", hs.Host, "server", srv.Name)
}
pipe(c, b, ss)
}
func dialAny(addrs []string, timeout time.Duration) (net.Conn, error) {
var errs []error
for _, a := range addrs {
conn, err := net.DialTimeout("tcp", a, timeout)
if err == nil {
return conn, nil
}
errs = append(errs, err)
}
return nil, errors.Join(errs...)
}
// pipe copies both ways and tears both connections down when either side ends.
func pipe(client, backend net.Conn, ss *serverStats) {
done := make(chan struct{}, 2)
go func() {
n, _ := io.Copy(backend, client)
ss.bytesIn.Add(n)
done <- struct{}{}
}()
go func() {
n, _ := io.Copy(client, backend)
ss.bytesOut.Add(n)
done <- struct{}{}
}()
<-done
client.Close()
backend.Close()
<-done
}
+234
View File
@@ -0,0 +1,234 @@
package proxy
import (
"bytes"
"context"
"encoding/binary"
"fmt"
"io"
"log/slog"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.veillax.com/Veillax/spinner/internal/config"
"git.veillax.com/Veillax/spinner/internal/health"
"git.veillax.com/Veillax/spinner/internal/protocol"
)
var quiet = slog.New(slog.NewTextHandler(io.Discard, nil))
// backend is a fake Minecraft server: status pings get a canned response, and
// login connections are answered with "hello from <name>" plus the PROXY header
// it saw, if any.
func backend(t *testing.T, name string, expectProxy bool) string {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
c, err := ln.Accept()
if err != nil {
return
}
go func() {
defer c.Close()
c.SetDeadline(time.Now().Add(5 * time.Second))
seen := "none"
if expectProxy {
hdr := make([]byte, 16)
if _, err := io.ReadFull(c, hdr); err != nil {
return
}
rest := make([]byte, binary.BigEndian.Uint16(hdr[14:]))
io.ReadFull(c, rest)
if bytes.Equal(hdr[:12], v2sig) && hdr[12] == 0x21 && hdr[13] == 0x11 {
seen = fmt.Sprintf("%d.%d.%d.%d", rest[0], rest[1], rest[2], rest[3])
}
}
hs, _, err := protocol.ReadHandshake(c)
if err != nil {
return
}
if hs.IsStatus() {
protocol.ServeStatus(c, `{"version":{"name":"fake","protocol":767},"players":{"online":2,"max":10},"description":{"text":"`+name+`"}}`)
return
}
fmt.Fprintf(c, "hello from %s proxy=%s host=%s", name, seen, hs.Host)
}()
}
}()
return ln.Addr().String()
}
func start(t *testing.T, cfgJSON string) (addr string, p *Proxy, h *health.Tracker) {
t.Helper()
path := filepath.Join(t.TempDir(), "spinner.json")
if err := os.WriteFile(path, []byte(cfgJSON), 0o600); err != nil {
t.Fatal(err)
}
m, err := config.NewManager(path, quiet)
if err != nil {
t.Fatal(err)
}
h = health.New(quiet)
p = New(m, h, quiet)
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
go p.Serve(ctx, ln)
return ln.Addr().String(), p, h
}
func login(t *testing.T, addr, host string) string {
t.Helper()
c, err := net.Dial("tcp", addr)
if err != nil {
t.Fatal(err)
}
defer c.Close()
c.SetDeadline(time.Now().Add(5 * time.Second))
c.Write(protocol.BuildHandshake(767, host, 25565, 2))
b, _ := io.ReadAll(c)
return string(b)
}
func status(t *testing.T, addr, host string) string {
t.Helper()
c, err := net.Dial("tcp", addr)
if err != nil {
t.Fatal(err)
}
defer c.Close()
js, err := protocol.QueryStatus(c, host, 25565, 5*time.Second)
if err != nil {
t.Fatal(err)
}
return js
}
func TestRoutesByHostname(t *testing.T) {
a, b := backend(t, "alpha", false), backend(t, "beta", false)
addr, _, _ := start(t, fmt.Sprintf(`{"servers":[
{"name":"alpha","domains":["alpha.test"],"addresses":[%q]},
{"name":"beta","domains":["*.beta.test"],"addresses":[%q]}]}`, a, b))
if got := login(t, addr, "alpha.test"); !strings.HasPrefix(got, "hello from alpha") {
t.Errorf("alpha: %q", got)
}
if got := login(t, addr, "x.beta.test"); !strings.HasPrefix(got, "hello from beta") || !strings.Contains(got, "host=x.beta.test") {
t.Errorf("beta: %q", got)
}
}
func TestProxyProtocolHeader(t *testing.T) {
a := backend(t, "pp", true)
addr, _, _ := start(t, fmt.Sprintf(`{"servers":[{"name":"pp","domains":["pp.test"],"addresses":[%q],"send_proxy_protocol":true}]}`, a))
if got := login(t, addr, "pp.test"); !strings.Contains(got, "proxy=127.0.0.1") {
t.Errorf("backend did not see client address in PROXY header: %q", got)
}
}
func TestFallsBackToSecondAddress(t *testing.T) {
good := backend(t, "good", false)
addr, _, _ := start(t, fmt.Sprintf(`{"servers":[{"name":"s","domains":["s.test"],"addresses":["127.0.0.1:1",%q]}]}`, good))
if got := login(t, addr, "s.test"); !strings.HasPrefix(got, "hello from good") {
t.Errorf("got %q", got)
}
}
func TestOfflineBackend(t *testing.T) {
addr, p, _ := start(t, `{"servers":[{"name":"dead","domains":["dead.test"],"addresses":["127.0.0.1:1"],"offline_motd":"napping"}]}`)
if js := status(t, addr, "dead.test"); !strings.Contains(js, "napping") {
t.Errorf("status: %s", js)
}
if got := login(t, addr, "dead.test"); !strings.Contains(got, "napping") {
t.Errorf("login: %q", got)
}
if f := p.Stats.Snapshot().Servers[0].Failures; f < 2 {
t.Errorf("failures = %d", f)
}
}
func TestUnknownHost(t *testing.T) {
addr, p, _ := start(t, `{"unknown_motd":"nobody home","servers":[]}`)
if js := status(t, addr, "who.test"); !strings.Contains(js, "nobody home") {
t.Errorf("status: %s", js)
}
if got := login(t, addr, "who.test"); !strings.Contains(got, "nobody home") {
t.Errorf("login: %q", got)
}
if p.Stats.Snapshot().Unknown != 2 {
t.Errorf("unknown = %d", p.Stats.Snapshot().Unknown)
}
}
func TestStatusServedFromHealthCache(t *testing.T) {
a := backend(t, "live", false)
addr, p, h := start(t, fmt.Sprintf(`{"probe_interval_secs":60,"servers":[{"name":"live","domains":["live.test"],"addresses":[%q]}]}`, a))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go h.Run(ctx, p.cfg)
deadline := time.Now().Add(3 * time.Second)
for {
if _, ok := h.Cached("live", time.Minute); ok {
break
}
if time.Now().After(deadline) {
t.Fatal("probe never succeeded")
}
time.Sleep(20 * time.Millisecond)
}
if js := status(t, addr, "live.test"); !strings.Contains(js, `"online":2`) {
t.Errorf("status: %s", js)
}
if snap := h.Snapshot(); len(snap) != 1 || !snap[0].Up || snap[0].Online != 2 || snap[0].Version != "fake" {
t.Errorf("snapshot: %+v", snap)
}
}
func TestPerIPLimit(t *testing.T) {
addr, p, _ := start(t, `{"max_conns_per_ip":1,"servers":[]}`)
hold, err := net.Dial("tcp", addr) // never sends a handshake, so it keeps its slot
if err != nil {
t.Fatal(err)
}
defer hold.Close()
time.Sleep(100 * time.Millisecond)
c, _ := net.Dial("tcp", addr)
defer c.Close()
c.SetDeadline(time.Now().Add(2 * time.Second))
if _, err := io.ReadAll(c); err != nil && !strings.Contains(err.Error(), "reset") {
t.Errorf("expected close, got %v", err)
}
if p.Stats.Snapshot().Rejected != 1 {
t.Errorf("rejected = %d", p.Stats.Snapshot().Rejected)
}
}
func TestConfigHotSwap(t *testing.T) {
a := backend(t, "one", false)
addr, p, _ := start(t, `{"servers":[]}`)
if got := login(t, addr, "one.test"); strings.HasPrefix(got, "hello") {
t.Fatalf("should be unknown: %q", got)
}
cfg, err := config.Parse([]byte(fmt.Sprintf(`{"servers":[{"name":"one","domains":["one.test"],"addresses":[%q]}]}`, a)))
if err != nil {
t.Fatal(err)
}
if err := p.cfg.Save(cfg); err != nil {
t.Fatal(err)
}
if got := login(t, addr, "one.test"); !strings.HasPrefix(got, "hello from one") {
t.Errorf("after save: %q", got)
}
}
+31
View File
@@ -0,0 +1,31 @@
package proxy
import (
"encoding/binary"
"net"
)
var v2sig = []byte("\r\n\r\n\x00\r\nQUIT\n")
// proxyV2Header builds a PROXY protocol v2 header carrying the client's real
// address so the backend can see it. Falls back to a LOCAL header if the
// addresses aren't TCP.
func proxyV2Header(src, dst net.Addr) []byte {
s, sok := src.(*net.TCPAddr)
d, dok := dst.(*net.TCPAddr)
h := append([]byte{}, v2sig...)
if !sok || !dok {
return append(h, 0x20, 0x00, 0, 0)
}
if s4, d4 := s.IP.To4(), d.IP.To4(); s4 != nil && d4 != nil {
h = append(h, 0x21, 0x11, 0, 12)
h = append(h, s4...)
h = append(h, d4...)
} else {
h = append(h, 0x21, 0x21, 0, 36)
h = append(h, s.IP.To16()...)
h = append(h, d.IP.To16()...)
}
h = binary.BigEndian.AppendUint16(h, uint16(s.Port))
return binary.BigEndian.AppendUint16(h, uint16(d.Port))
}
+75
View File
@@ -0,0 +1,75 @@
package proxy
import (
"sort"
"sync"
"sync/atomic"
"time"
)
type serverStats struct {
active atomic.Int64
total atomic.Int64
failures atomic.Int64
bytesIn atomic.Int64 // client -> backend
bytesOut atomic.Int64 // backend -> client
}
type Stats struct {
Started time.Time
active atomic.Int64
total atomic.Int64
rejected atomic.Int64
unknown atomic.Int64
mu sync.Mutex
servers map[string]*serverStats
}
func newStats() *Stats {
return &Stats{Started: time.Now(), servers: map[string]*serverStats{}}
}
func (s *Stats) server(name string) *serverStats {
s.mu.Lock()
defer s.mu.Unlock()
ss := s.servers[name]
if ss == nil {
ss = &serverStats{}
s.servers[name] = ss
}
return ss
}
type ServerSnapshot struct {
Name string `json:"name"`
Active int64 `json:"active"`
Total int64 `json:"total"`
Failures int64 `json:"failures"`
BytesIn int64 `json:"bytes_in"`
BytesOut int64 `json:"bytes_out"`
}
type Snapshot struct {
UptimeSecs int64 `json:"uptime_secs"`
Active int64 `json:"active"`
Total int64 `json:"total"`
Rejected int64 `json:"rejected"`
Unknown int64 `json:"unknown_host"`
Servers []ServerSnapshot `json:"servers"`
}
func (s *Stats) Snapshot() Snapshot {
out := Snapshot{
UptimeSecs: int64(time.Since(s.Started).Seconds()),
Active: s.active.Load(), Total: s.total.Load(),
Rejected: s.rejected.Load(), Unknown: s.unknown.Load(),
}
s.mu.Lock()
for n, ss := range s.servers {
out.Servers = append(out.Servers, ServerSnapshot{n, ss.active.Load(), ss.total.Load(),
ss.failures.Load(), ss.bytesIn.Load(), ss.bytesOut.Load()})
}
s.mu.Unlock()
sort.Slice(out.Servers, func(i, j int) bool { return out.Servers[i].Name < out.Servers[j].Name })
return out
}
+60
View File
@@ -0,0 +1,60 @@
// Package router maps the hostname a client typed to a configured server.
package router
import (
"sort"
"strings"
"git.veillax.com/Veillax/spinner/internal/config"
)
type Router struct {
exact map[string]*config.Server
wild []wildcard
}
type wildcard struct {
suffix string // ".example.com"
srv *config.Server
}
// Normalize lowercases a handshake host and strips the trailing dot and any
// Forge "\0FML\0" suffix.
func Normalize(host string) string {
if i := strings.IndexByte(host, 0); i >= 0 {
host = host[:i]
}
return strings.ToLower(strings.TrimRight(host, "."))
}
// New builds a router. Exact domains win; wildcards (*.example.com) match any
// depth of subdomain but not the apex, and longer suffixes win over shorter.
func New(cfg *config.Config) *Router {
r := &Router{exact: map[string]*config.Server{}}
for i := range cfg.Servers {
s := &cfg.Servers[i]
for _, d := range s.Domains {
d = strings.ToLower(strings.TrimRight(d, "."))
if rest, ok := strings.CutPrefix(d, "*."); ok {
r.wild = append(r.wild, wildcard{"." + rest, s})
} else if _, dup := r.exact[d]; !dup {
r.exact[d] = s
}
}
}
sort.SliceStable(r.wild, func(i, j int) bool { return len(r.wild[i].suffix) > len(r.wild[j].suffix) })
return r
}
func (r *Router) Match(host string) *config.Server {
host = Normalize(host)
if s, ok := r.exact[host]; ok {
return s
}
for _, w := range r.wild {
if strings.HasSuffix(host, w.suffix) && len(host) > len(w.suffix) {
return w.srv
}
}
return nil
}
+55
View File
@@ -0,0 +1,55 @@
package router
import (
"testing"
"git.veillax.com/Veillax/spinner/internal/config"
)
func TestMatch(t *testing.T) {
cfg, err := config.Parse([]byte(`{"servers":[
{"name":"a","domains":["a.example.com","*.deep.example.com"],"addresses":["127.0.0.1:1"]},
{"name":"b","domains":["*.example.com"],"addresses":["127.0.0.1:2"]},
{"name":"c","domains":["*.x.deep.example.com"],"addresses":["127.0.0.1:3"]}
]}`))
if err != nil {
t.Fatal(err)
}
r := New(cfg)
cases := map[string]string{
"a.example.com": "a",
"A.Example.COM.": "a",
"a.example.com\x00FML\x00": "a",
"q.deep.example.com": "a",
"other.example.com": "b",
"s.x.deep.example.com": "c",
"example.com": "",
"evil.com": "",
"notexample.com": "",
"": "",
}
for host, want := range cases {
got := ""
if s := r.Match(host); s != nil {
got = s.Name
}
if got != want {
t.Errorf("Match(%q) = %q, want %q", host, got, want)
}
}
}
func TestConfigValidation(t *testing.T) {
bad := []string{
`{"servers":[{"name":"","domains":["a"],"addresses":["h:1"]}]}`,
`{"servers":[{"name":"a","domains":[],"addresses":["h:1"]}]}`,
`{"servers":[{"name":"a","domains":["a"],"addresses":["nope"]}]}`,
`{"servers":[{"name":"a","domains":["a"],"addresses":["h:1"]},{"name":"a","domains":["b"],"addresses":["h:1"]}]}`,
`{"unknown_field":1}`,
}
for _, b := range bad {
if _, err := config.Parse([]byte(b)); err == nil {
t.Errorf("expected error for %s", b)
}
}
}