Multi-host phase 1: API tokens, auto TLS with pinned fingerprints, hub-side node manager + reverse proxy (WebSockets included), machine switcher in the rail, Machines page
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Mitchell-Johnson
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
fc42675e9f
commit
7dcdb5dc98
@@ -0,0 +1,323 @@
|
||||
// Package nodes is the hub side of multi-host: the list of managed Ham Labs, their
|
||||
// health, adding one (pair with its password, receive a token, pin its certificate),
|
||||
// removing one (revoke the token), and the reverse proxy that turns
|
||||
// /api/nodes/<name>/… into /api/… on that node, WebSockets included.
|
||||
package nodes
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"hamlab/internal/config"
|
||||
"hamlab/internal/tlsauto"
|
||||
)
|
||||
|
||||
var nameRe = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
|
||||
|
||||
type Status struct {
|
||||
Name string `json:"name"`
|
||||
URL string `json:"url"`
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
Added time.Time `json:"added"`
|
||||
Online bool `json:"online"`
|
||||
Host string `json:"host,omitempty"`
|
||||
Version string `json:"version,omitempty"`
|
||||
Load string `json:"load,omitempty"`
|
||||
Uptime float64 `json:"uptime,omitempty"`
|
||||
RebootNeed bool `json:"reboot_needed,omitempty"`
|
||||
LastSeen time.Time `json:"last_seen,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
cfg *config.Config
|
||||
hubName string
|
||||
mu sync.Mutex
|
||||
status map[string]*Status
|
||||
proxies map[string]*httputil.ReverseProxy
|
||||
clients map[string]*http.Client
|
||||
}
|
||||
|
||||
func New(cfg *config.Config, hubName string) *Manager {
|
||||
return &Manager{cfg: cfg, hubName: hubName, status: map[string]*Status{}, proxies: map[string]*httputil.ReverseProxy{}, clients: map[string]*http.Client{}}
|
||||
}
|
||||
|
||||
func transportFor(n config.Node) *http.Transport {
|
||||
t := &http.Transport{
|
||||
DialContext: (&net.Dialer{Timeout: 5 * time.Second}).DialContext,
|
||||
TLSHandshakeTimeout: 5 * time.Second,
|
||||
ResponseHeaderTimeout: 60 * time.Second,
|
||||
MaxIdleConnsPerHost: 8,
|
||||
}
|
||||
if strings.HasPrefix(n.URL, "https://") {
|
||||
if n.Fingerprint != "" {
|
||||
t.TLSClientConfig = tlsauto.PinnedConfig(n.Fingerprint)
|
||||
} else {
|
||||
t.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
|
||||
}
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
func (m *Manager) client(n config.Node) *http.Client {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if c, ok := m.clients[n.Name]; ok {
|
||||
return c
|
||||
}
|
||||
c := &http.Client{Transport: transportFor(n), Timeout: 15 * time.Second}
|
||||
m.clients[n.Name] = c
|
||||
return c
|
||||
}
|
||||
|
||||
func (m *Manager) node(name string) (config.Node, bool) {
|
||||
var found config.Node
|
||||
ok := false
|
||||
m.cfg.View(func(c *config.Config) {
|
||||
for _, n := range c.Nodes {
|
||||
if n.Name == name {
|
||||
found, ok = n, true
|
||||
}
|
||||
}
|
||||
})
|
||||
return found, ok
|
||||
}
|
||||
|
||||
// ---- probing ----
|
||||
|
||||
func (m *Manager) probe(ctx context.Context, n config.Node) *Status {
|
||||
st := &Status{Name: n.Name, URL: n.URL, Fingerprint: n.Fingerprint, Added: n.Added}
|
||||
req, _ := http.NewRequestWithContext(ctx, "GET", strings.TrimRight(n.URL, "/")+"/api/me", nil)
|
||||
req.Header.Set("Authorization", "Bearer "+n.Token)
|
||||
resp, err := m.client(n).Do(req)
|
||||
if err != nil {
|
||||
st.Error = shortErr(err)
|
||||
return st
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != 200 {
|
||||
st.Error = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||||
if resp.StatusCode == 401 {
|
||||
st.Error = "token rejected (re-add the machine)"
|
||||
}
|
||||
return st
|
||||
}
|
||||
var me struct {
|
||||
Host, Version, Load string
|
||||
Uptime float64
|
||||
RebootNeeded bool `json:"reboot_needed"`
|
||||
}
|
||||
if json.NewDecoder(io.LimitReader(resp.Body, 64<<10)).Decode(&me) != nil {
|
||||
st.Error = "bad reply"
|
||||
return st
|
||||
}
|
||||
st.Online, st.Host, st.Version, st.Load, st.Uptime, st.RebootNeed, st.LastSeen = true, me.Host, me.Version, me.Load, me.Uptime, me.RebootNeeded, time.Now()
|
||||
return st
|
||||
}
|
||||
|
||||
func shortErr(err error) string {
|
||||
s := err.Error()
|
||||
if i := strings.LastIndex(s, ": "); i >= 0 {
|
||||
s = s[i+2:]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// Run refreshes every node's status periodically.
|
||||
func (m *Manager) Run(ctx context.Context) {
|
||||
m.refresh(ctx)
|
||||
t := time.NewTicker(15 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
m.refresh(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) refresh(ctx context.Context) {
|
||||
var list []config.Node
|
||||
m.cfg.View(func(c *config.Config) { list = append(list, c.Nodes...) })
|
||||
var wg sync.WaitGroup
|
||||
for _, n := range list {
|
||||
wg.Add(1)
|
||||
go func(n config.Node) {
|
||||
defer wg.Done()
|
||||
pctx, cancel := context.WithTimeout(ctx, 8*time.Second)
|
||||
defer cancel()
|
||||
st := m.probe(pctx, n)
|
||||
m.mu.Lock()
|
||||
if prev, ok := m.status[n.Name]; ok && !st.Online && prev.Online {
|
||||
st.LastSeen = prev.LastSeen // keep the last good sighting
|
||||
}
|
||||
m.status[n.Name] = st
|
||||
m.mu.Unlock()
|
||||
}(n)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// List returns every managed node with its last known status.
|
||||
func (m *Manager) List() []Status {
|
||||
var list []config.Node
|
||||
m.cfg.View(func(c *config.Config) { list = append(list, c.Nodes...) })
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
out := make([]Status, 0, len(list))
|
||||
for _, n := range list {
|
||||
if st, ok := m.status[n.Name]; ok {
|
||||
out = append(out, *st)
|
||||
} else {
|
||||
out = append(out, Status{Name: n.Name, URL: n.URL, Fingerprint: n.Fingerprint, Added: n.Added, Error: "not probed yet"})
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
||||
return out
|
||||
}
|
||||
|
||||
// ---- add / remove ----
|
||||
|
||||
// Add pairs with a node: pin its certificate (trust on first use), trade the node's
|
||||
// password for a token named after this hub, and store the node. The password is
|
||||
// used once and never kept.
|
||||
func (m *Manager) Add(ctx context.Context, name, rawURL, password string) (Status, error) {
|
||||
if !nameRe.MatchString(name) {
|
||||
return Status{}, errors.New("name: lowercase letters, digits and dashes only")
|
||||
}
|
||||
if _, exists := m.node(name); exists {
|
||||
return Status{}, errors.New("a machine with that name already exists")
|
||||
}
|
||||
u, err := url.Parse(strings.TrimSpace(rawURL))
|
||||
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
|
||||
return Status{}, errors.New("address must be http(s)://host:port")
|
||||
}
|
||||
n := config.Node{Name: name, URL: u.Scheme + "://" + u.Host, Added: time.Now()}
|
||||
if u.Scheme == "https" {
|
||||
host := u.Host
|
||||
if _, _, err := net.SplitHostPort(host); err != nil {
|
||||
host += ":443"
|
||||
}
|
||||
fp, err := tlsauto.Peek(host)
|
||||
if err != nil {
|
||||
return Status{}, fmt.Errorf("connect: %s", shortErr(err))
|
||||
}
|
||||
n.Fingerprint = fp
|
||||
}
|
||||
body, _ := json.Marshal(map[string]string{"password": password, "name": "hub:" + m.hubName})
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", n.URL+"/api/tokens", bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := (&http.Client{Transport: transportFor(n), Timeout: 15 * time.Second}).Do(req)
|
||||
if err != nil {
|
||||
return Status{}, fmt.Errorf("connect: %s", shortErr(err))
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var reply struct {
|
||||
Token string `json:"token"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
_ = json.NewDecoder(io.LimitReader(resp.Body, 64<<10)).Decode(&reply)
|
||||
if resp.StatusCode != 200 || reply.Token == "" {
|
||||
if reply.Error == "" {
|
||||
reply.Error = fmt.Sprintf("HTTP %d", resp.StatusCode)
|
||||
}
|
||||
return Status{}, errors.New("pairing refused: " + reply.Error)
|
||||
}
|
||||
n.Token = reply.Token
|
||||
if err := m.cfg.Update(func(c *config.Config) { c.Nodes = append(c.Nodes, n) }); err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
st := m.probe(ctx, n)
|
||||
m.mu.Lock()
|
||||
m.status[name] = st
|
||||
m.mu.Unlock()
|
||||
return *st, nil
|
||||
}
|
||||
|
||||
// Remove revokes our token on the node (best effort) and forgets it.
|
||||
func (m *Manager) Remove(ctx context.Context, name string) error {
|
||||
n, ok := m.node(name)
|
||||
if !ok {
|
||||
return errors.New("no such machine")
|
||||
}
|
||||
req, _ := http.NewRequestWithContext(ctx, "DELETE", n.URL+"/api/tokens/"+url.PathEscape("hub:"+m.hubName), nil)
|
||||
req.Header.Set("Authorization", "Bearer "+n.Token)
|
||||
if resp, err := m.client(n).Do(req); err == nil {
|
||||
resp.Body.Close()
|
||||
}
|
||||
err := m.cfg.Update(func(c *config.Config) {
|
||||
kept := c.Nodes[:0]
|
||||
for _, x := range c.Nodes {
|
||||
if x.Name != name {
|
||||
kept = append(kept, x)
|
||||
}
|
||||
}
|
||||
c.Nodes = kept
|
||||
})
|
||||
m.mu.Lock()
|
||||
delete(m.status, name)
|
||||
delete(m.proxies, name)
|
||||
delete(m.clients, name)
|
||||
m.mu.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
// ---- proxy ----
|
||||
|
||||
// Proxy returns the handler for /api/nodes/<name>/<rest>: it rewrites to <node>/api/<rest>,
|
||||
// swaps the hub's cookie for the node's bearer token, and drops Origin so the node's
|
||||
// WebSocket origin check (meant for browsers) doesn't reject the hub.
|
||||
func (m *Manager) Proxy(name string) (http.Handler, bool) {
|
||||
n, ok := m.node(name)
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
m.mu.Lock()
|
||||
if p, ok := m.proxies[name]; ok {
|
||||
m.mu.Unlock()
|
||||
return p, true
|
||||
}
|
||||
m.mu.Unlock()
|
||||
target, _ := url.Parse(n.URL)
|
||||
prefix := "/api/nodes/" + name
|
||||
p := &httputil.ReverseProxy{
|
||||
Director: func(r *http.Request) {
|
||||
r.URL.Scheme, r.URL.Host = target.Scheme, target.Host
|
||||
r.URL.Path = "/api" + strings.TrimPrefix(r.URL.Path, prefix)
|
||||
r.URL.RawPath = ""
|
||||
r.Host = target.Host
|
||||
r.Header.Set("Authorization", "Bearer "+n.Token)
|
||||
r.Header.Del("Cookie")
|
||||
r.Header.Del("Origin")
|
||||
r.Header.Del("X-Forwarded-For")
|
||||
r.Header.Del("X-Forwarded-Proto")
|
||||
},
|
||||
Transport: transportFor(n),
|
||||
FlushInterval: -1, // stream task output / log tails immediately
|
||||
ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusBadGateway)
|
||||
_ = json.NewEncoder(w).Encode(map[string]string{"error": name + " unreachable: " + shortErr(err)})
|
||||
},
|
||||
}
|
||||
m.mu.Lock()
|
||||
m.proxies[name] = p
|
||||
m.mu.Unlock()
|
||||
return p, true
|
||||
}
|
||||
Reference in New Issue
Block a user