Hammerhead/store.go

// Package hammerhead is the control plane: JSON API, web UI, accounts and state store.
package hammerhead

import (
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"git.veillax.com/Veillax/reef/model"
	"git.veillax.com/Veillax/reef/secretbox"
	"unicode"
	"unicode/utf8"
)

type Org struct {
	ID         string `json:"id"`
	Name       string `json:"name"`
	AllowLocal bool   `json:"allowLocal"` // may target an edge's built-in local site (the edge machine itself)

	AlertDebounceMin int `json:"alertDebounceMin,omitempty"` // minutes a problem must last before alerting; 0 = 2
	AlertRepeatHours int `json:"alertRepeatHours,omitempty"` // remind about an unresolved problem this often; 0 = never

	// Traffic alerts, applied to every port of the organization. 0 = off.
	TrafficSpikeFactor int `json:"trafficSpikeFactor,omitempty"` // alert when a port carries this many times its usual traffic
	TrafficDailyMB     int `json:"trafficDailyMb,omitempty"`     // alert when a port carries more than this many MB in one UTC day
}

type Role struct {
	ID      string `json:"id"`
	OrgID   string `json:"orgId"`
	Name    string `json:"name"`
	Admin   bool   `json:"admin"`
	Builtin bool   `json:"builtin"`
}

type Member struct {
	UserID string `json:"userId"`
	OrgID  string `json:"orgId"`
	RoleID string `json:"roleId"`
}

type User struct {
	ID          string    `json:"id"`
	Email       string    `json:"email"`
	Name        string    `json:"name"`
	PassHash    string    `json:"passHash,omitempty"`
	ServerAdmin bool      `json:"serverAdmin"`
	TOTPSecret  string    `json:"totpSecret,omitempty"`
	TOTPEnabled bool      `json:"totpEnabled"`
	BackupCodes []string  `json:"backupCodes,omitempty"` // sha256 hashes
	IdP         string    `json:"idp,omitempty"`
	IdPSub      string    `json:"idpSub,omitempty"`
	Created     time.Time `json:"created"`
}

type Session struct {
	UA      string    `json:"ua,omitempty"`
	UserID  string    `json:"userId"`
	Expires time.Time `json:"expires"`
	Created time.Time `json:"created"`
	IP      string    `json:"ip"`
}

type Invite struct {
	ID        string    `json:"id"`
	OrgID     string    `json:"orgId"`
	Email     string    `json:"email"`
	RoleID    string    `json:"roleId"`
	TokenHash string    `json:"tokenHash"`
	Expires   time.Time `json:"expires"`
	Created   time.Time `json:"created"`
}

type APIKey struct {
	ID       string    `json:"id"`
	OrgID    string    `json:"orgId"`
	Name     string    `json:"name"`
	Hash     string    `json:"hash"`
	Created  time.Time `json:"created"`
	LastUsed time.Time `json:"lastUsed"`
}

type IdP struct {
	ID             string   `json:"id"`
	Name           string   `json:"name"`
	Issuer         string   `json:"issuer"`
	ClientID       string   `json:"clientId"`
	ClientSecret   string   `json:"clientSecret,omitempty"`
	Scopes         string   `json:"scopes"` // default "openid email profile"
	AutoProvision  bool     `json:"autoProvision"`
	OrgID          string   `json:"orgId"`  // org new users are added to
	RoleID         string   `json:"roleId"` // ...with this role
	AllowedDomains []string `json:"allowedDomains"`
}

type SMTP struct {
	Host string `json:"host"`
	Port int    `json:"port"`
	User string `json:"user"`
	Pass string `json:"pass,omitempty"`
	From string `json:"from"`
	Mode string `json:"mode"` // starttls, tls, none
}

type Settings struct {
	DashboardURL string      `json:"dashboardUrl"`
	AcmeEmail    string      `json:"acmeEmail"`
	AcmeStaging  bool        `json:"acmeStaging"`
	SMTP         SMTP        `json:"smtp"`
	Sandbar      SandbarSync `json:"sandbar"`
	// SelfEdgeID is this server's own edge (see HAMMERHEAD_SELF_DOMAIN).
	SelfEdgeID string `json:"selfEdgeId,omitempty"`
	// Terminal holds the web terminal's SSH key and the host keys it has seen (see terminal.go).
	Terminal TerminalSettings `json:"terminal"`
}

type Secret struct {
	PasswordHash string `json:"passwordHash,omitempty"`
	PinHash      string `json:"pinHash,omitempty"`
}

type ShareLink struct {
	ID        string    `json:"id"`
	MappingID string    `json:"mappingId"`
	OrgID     string    `json:"orgId"`
	Name      string    `json:"name"`
	TokenHash string    `json:"tokenHash"`
	Expires   time.Time `json:"expires"` // zero = never
	MaxUses   int       `json:"maxUses"` // 0 = unlimited
	Uses      int       `json:"uses"`
	Created   time.Time `json:"created"`
}

type Total struct {
	Conns    int64 `json:"conns"`
	BytesIn  int64 `json:"bytesIn"`
	BytesOut int64 `json:"bytesOut"`
}

type state struct {
	Version    int64               `json:"version"`
	Settings   Settings            `json:"settings"`
	Users      []*User             `json:"users"`
	Sessions   map[string]*Session `json:"sessions"` // by sha256(token)
	Orgs       []*Org              `json:"orgs"`
	Roles      []*Role             `json:"roles"`
	Members    []*Member           `json:"members"`
	Invites    []*Invite           `json:"invites"`
	APIKeys    []*APIKey           `json:"apiKeys"`
	IdPs       []*IdP              `json:"idps"`
	Edges      []*model.Edge       `json:"edges"`
	Sites      []*model.Site       `json:"sites"`
	Domains    []*model.Domain     `json:"domains"`
	Mappings   []*model.Mapping    `json:"mappings"`
	MailEgress []*model.MailEgress `json:"mailEgress"`
	DKIM       []*model.DKIMKey    `json:"dkim"`
	Channels   []*AlertChannel     `json:"channels"`
	Secrets    map[string]*Secret  `json:"secrets"` // by mapping ID
	Shares     []*ShareLink        `json:"shares"`
	Totals     map[string]*Total   `json:"totals"` // by mapping ID
}

type Options struct {
	AdminEmail    string // optional: create this server admin on first run
	AdminPassword string
	DashboardURL  string // optional: overrides the saved setting

	MasterKey        string        // encrypts secrets in the database (HAMMERHEAD_MASTER_KEY); generated into <data>/master.key if empty
	MetricsToken     string        // enables GET /metrics for a bearer token (HAMMERHEAD_METRICS_TOKEN)
	LogRetentionDays int           // access logs are kept this long; 0 = 30
	AlertInterval    time.Duration // how often alerts are evaluated; 0 = 15s
	AlertDebounce    time.Duration // overrides the per-organization delay before alerting (tests)
	OfflineAfter     time.Duration // how long an edge or site may stay silent before it counts as down; 0 = 90s

	// SelfDomain (HAMMERHEAD_SELF_DOMAIN) gives this server an edge of its own, with a built-in
	// https route for the dashboard at that name, so no reverse proxy is needed in front of it.
	// SelfTarget is the address Hammerhead listens on (e.g. 127.0.0.1:8080); SelfTunnel is the
	// edge's public WireGuard endpoint (default <SelfDomain>:51820).
	SelfDomain string
	SelfTarget string
	SelfTunnel string

	// Version is the running release (e.g. v0.1.5), shown under Settings with a check for updates.
	Version string
	// AutoUpdate says whether this process updates itself (HAMMERHEAD_AUTO_UPDATE), for the same screen.
	AutoUpdate bool
	// UpdateServer and UpdateRepo say where releases are published; the defaults are this project's own.
	UpdateServer string
	UpdateRepo   string
}

type Server struct {
	mu   sync.Mutex
	dir  string
	st   state
	opts Options

	status     map[string]model.MappingStatus
	mailStatus map[string]string                    // mail egress ID -> "" (listening) or an error
	mailChecks map[string]model.MailCheck           // mail egress ID -> the edge's deliverability report
	containers map[string][]model.Container         // by site ID
	discovered map[string][]model.DiscoveredService // by site ID: suggested targets the agent found
	terms      map[string]*pendingTerm              // terminal sessions waiting for an agent to call back, by session ID
	changed    chan struct{}

	tickets map[string]ticket
	oauth   map[string]oauthState
	otps    map[string]otpEntry
	limits  map[string][]time.Time
	oidc    map[string]*oidcDoc

	alerts   map[string]*alertState // active problems by key (see alerts.go)
	updRuns  map[string]*updRun     // progress of updates running on edges and sites, by ID
	selfUpd  selfUpdate             // progress of this server's own update
	verMu    sync.Mutex             // guards verCache and verAt, the latest releases seen by /api/versions
	verCache map[string]string
	verAt    time.Time
	mtaSent  map[string]time.Time // reminder throttle for repeat notifications

	expose map[string]time.Time // deadline of each ephemeral mapping, by ID (see expose.go)

	sandbarMu     sync.Mutex        // one monitor sync at a time (see sandbar.go)
	sandbarStatus sandbarSyncStatus // how the last one went

	logs *logStore
	db   *DB
	stop chan struct{}
	once sync.Once

	// Mailer sends email; tests replace it.
	Mailer func(to, subject, body string) error
}

func New(dir string, opts Options) (*Server, error) {
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return nil, err
	}
	box, err := masterBox(dir, opts.MasterKey)
	if err != nil {
		return nil, err
	}
	db, err := openDB(filepath.Join(dir, "hammerhead.db"), box)
	if err != nil {
		return nil, err
	}
	s := &Server{
		dir:        dir,
		opts:       opts,
		db:         db,
		logs:       &logStore{db: db},
		stop:       make(chan struct{}),
		status:     map[string]model.MappingStatus{},
		mailStatus: map[string]string{},
		mailChecks: map[string]model.MailCheck{},
		containers: map[string][]model.Container{},
		discovered: map[string][]model.DiscoveredService{},
		terms:      map[string]*pendingTerm{},
		changed:    make(chan struct{}),
		tickets:    map[string]ticket{},
		oauth:      map[string]oauthState{},
		otps:       map[string]otpEntry{},
		limits:     map[string][]time.Time{},
		oidc:       map[string]*oidcDoc{},
		alerts:     map[string]*alertState{},
		mtaSent:    map[string]time.Time{},
		updRuns:    map[string]*updRun{},
		expose:     map[string]time.Time{},
	}
	s.st.Sessions = map[string]*Session{}
	s.st.Secrets = map[string]*Secret{}
	s.st.Totals = map[string]*Total{}
	had, err := db.load(&s.st)
	if err != nil {
		db.Close()
		return nil, err
	}
	if !had {
		if err := s.migrateLegacy(); err != nil {
			db.Close()
			return nil, err
		}
	}
	if opts.DashboardURL != "" {
		s.st.Settings.DashboardURL = strings.TrimRight(opts.DashboardURL, "/")
	}
	s.backfillTunnelIPs()
	s.Mailer = s.smtpSend
	if len(s.st.Users) == 0 && opts.AdminEmail != "" && opts.AdminPassword != "" {
		if err := s.createFirstAdmin(opts.AdminEmail, "Admin", opts.AdminPassword, "Default"); err != nil {
			db.Close()
			return nil, err
		}
	}
	if opts.SelfDomain != "" && opts.SelfTunnel == "" {
		s.opts.SelfTunnel = opts.SelfDomain + ":51820"
	}
	s.finishSelfUpdateLog()
	if err := s.ensureSelfEdge(); err != nil {
		db.Close()
		return nil, err
	}
	if err := s.save(); err != nil {
		db.Close()
		return nil, err
	}
	s.loadAlerts()
	go s.housekeeping()
	go s.exposeLoop()
	go s.alertLoop()
	go s.sandbarLoop()
	return s, nil
}

// masterBox builds the encryption box for secrets at rest. The key normally
// comes from HAMMERHEAD_MASTER_KEY (e.g. in .env). If none is given one is
// generated into <data>/master.key so the server works out of the box, but that
// file sits next to the database, so a backup of the folder carries both: keep
// the key somewhere else for real protection.
func masterBox(dir, key string) (*secretbox.Box, error) {
	if key != "" {
		return secretbox.New(key)
	}
	path := filepath.Join(dir, "master.key")
	if b, err := os.ReadFile(path); err == nil {
		return secretbox.New(string(b))
	}
	k := secretbox.GenerateKey()
	if err := os.WriteFile(path, []byte(k+"\n"), 0o600); err != nil {
		return nil, err
	}
	log.Printf("no HAMMERHEAD_MASTER_KEY set: generated an encryption key in %s. For real protection copy its contents into your .env as HAMMERHEAD_MASTER_KEY=... and delete the file, so a backup of the data folder doesn't carry the key too", path)
	return secretbox.New(k)
}

// migrateLegacy imports a state.json (and JSONL logs) from before the database existed.
func (s *Server) migrateLegacy() error {
	path := filepath.Join(s.dir, "state.json")
	b, err := os.ReadFile(path)
	if err != nil {
		return nil // a fresh install
	}
	if err := json.Unmarshal(b, &s.st); err != nil {
		return fmt.Errorf("import %s: %w", path, err)
	}
	if s.st.Sessions == nil {
		s.st.Sessions = map[string]*Session{}
	}
	if s.st.Secrets == nil {
		s.st.Secrets = map[string]*Secret{}
	}
	if s.st.Totals == nil {
		s.st.Totals = map[string]*Total{}
	}
	if err := s.db.persist(&s.st); err != nil {
		return err
	}
	s.importLegacyLogs()
	os.Rename(path, path+".migrated")
	log.Printf("imported %s into the database (kept as %s.migrated)", path, path)
	return nil
}

// Close stops background work and closes the database.
func (s *Server) Close() {
	s.once.Do(func() {
		close(s.stop)
		s.db.Close()
	})
}

// housekeeping prunes old logs.
func (s *Server) housekeeping() {
	days := s.opts.LogRetentionDays
	if days <= 0 {
		days = 30
	}
	t := time.NewTicker(time.Hour)
	defer t.Stop()
	for {
		s.db.purge(days, 365)
		select {
		case <-s.stop:
			return
		case <-t.C:
		}
	}
}

// save must be called with s.mu held (or before serving).
func (s *Server) save() error {
	err := s.db.persist(&s.st)
	if err != nil {
		log.Printf("save: %v", err)
	}
	return err
}

// bump records a change that agents care about and wakes every long-poller. Caller holds s.mu.
func (s *Server) bump() {
	s.st.Version++
	s.save()
	close(s.changed)
	s.changed = make(chan struct{})
}

// wait blocks until the version moves past have, or 25s pass. It returns false on timeout.
func (s *Server) wait(r *http.Request, have int64) bool { return s.waitUntil(r, have, nil) }

// waitUntil is wait that also returns at once when ready (called with s.mu held) says there is
// something to deliver that does not move the version, such as a terminal waiting for an agent.
// Checking it under the same lock that takes the change channel is what keeps a request that
// arrives between two polls from being missed.
func (s *Server) waitUntil(r *http.Request, have int64, ready func() bool) bool {
	s.mu.Lock()
	if s.st.Version > have || (ready != nil && ready()) {
		s.mu.Unlock()
		return true
	}
	ch := s.changed
	s.mu.Unlock()
	select {
	case <-ch:
		return true
	case <-time.After(25 * time.Second):
		return false
	case <-r.Context().Done():
		return false
	}
}

// ---------- small helpers ----------

func randHex(n int) string {
	b := make([]byte, n)
	rand.Read(b)
	return hex.EncodeToString(b)
}

func newID() string { return randHex(6) }

func sha(s string) string {
	h := sha256.Sum256([]byte(s))
	return hex.EncodeToString(h[:])
}

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(v)
}

// fail sends an error to the browser. Messages are shown to people, so they start with a capital letter.
func fail(w http.ResponseWriter, code int, err error) {
	msg := err.Error()
	if r, n := utf8.DecodeRuneInString(msg); n > 0 {
		msg = string(unicode.ToUpper(r)) + msg[n:]
	}
	writeJSON(w, code, map[string]string{"error": msg})
}

func failf(w http.ResponseWriter, code int, format string, a ...any) {
	fail(w, code, fmt.Errorf(format, a...))
}

func ok(w http.ResponseWriter) { writeJSON(w, 200, map[string]bool{"ok": true}) }

func decode(w http.ResponseWriter, r *http.Request, v any) bool {
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(v); err != nil {
		fail(w, 400, errors.New("invalid request body"))
		return false
	}
	return true
}

func filter[T any](in []*T, keep func(*T) bool) []*T {
	out := in[:0:0]
	for _, v := range in {
		if keep(v) {
			out = append(out, v)
		}
	}
	return out
}

func bearer(r *http.Request) string {
	return strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
}

func (s *Server) dashboardURL() string { return s.st.Settings.DashboardURL }

// allow is a sliding-window rate limiter. Caller must NOT hold s.mu.
func (s *Server) allow(key string, max int, window time.Duration) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := time.Now()
	kept := s.limits[key][:0]
	for _, t := range s.limits[key] {
		if now.Sub(t) < window {
			kept = append(kept, t)
		}
	}
	if len(kept) >= max {
		s.limits[key] = kept
		return false
	}
	s.limits[key] = append(kept, now)
	return true
}

// ---------- lookups (caller holds s.mu) ----------

func (s *Server) findOrg(id string) *Org {
	for _, o := range s.st.Orgs {
		if o.ID == id {
			return o
		}
	}
	return nil
}

func (s *Server) findRole(id string) *Role {
	for _, r := range s.st.Roles {
		if r.ID == id {
			return r
		}
	}
	return nil
}

func (s *Server) findUser(id string) *User {
	for _, u := range s.st.Users {
		if u.ID == id {
			return u
		}
	}
	return nil
}

func (s *Server) userByEmail(email string) *User {
	email = strings.ToLower(strings.TrimSpace(email))
	for _, u := range s.st.Users {
		if strings.ToLower(u.Email) == email {
			return u
		}
	}
	return nil
}

func (s *Server) findEdge(id string) *model.Edge {
	for _, e := range s.st.Edges {
		if e.ID == id {
			return e
		}
	}
	return nil
}

func (s *Server) findSite(id string) *model.Site {
	for _, e := range s.st.Sites {
		if e.ID == id {
			return e
		}
	}
	return nil
}

func (s *Server) findMapping(id string) *model.Mapping {
	for _, m := range s.st.Mappings {
		if m.ID == id {
			return m
		}
	}
	return nil
}

func (s *Server) member(userID, orgID string) *Member {
	for _, m := range s.st.Members {
		if m.UserID == userID && m.OrgID == orgID {
			return m
		}
	}
	return nil
}

// allocIPLocked picks the next free address in the tunnel network (100.89.0.0/16).
// The network only exists inside WireGuard's userspace stack, so it can never
// clash with a LAN or VPN on the machines involved. Caller holds s.mu.
func (s *Server) allocIPLocked() string {
	used := map[string]bool{}
	for _, e := range s.st.Edges {
		used[e.TunnelIP] = true
	}
	for _, st := range s.st.Sites {
		used[st.TunnelIP] = true
	}
	for i := 1; i < 65535; i++ {
		ip := fmt.Sprintf("100.89.%d.%d", i>>8, i&0xff)
		if i&0xff != 0 && i&0xff != 255 && !used[ip] {
			return ip
		}
	}
	return ""
}

// backfillTunnelIPs gives addresses to records created before WireGuard existed.
func (s *Server) backfillTunnelIPs() {
	for _, e := range s.st.Edges {
		if e.TunnelIP == "" {
			e.TunnelIP = s.allocIPLocked()
		}
	}
	for _, st := range s.st.Sites {
		if !st.Local && st.TunnelIP == "" {
			st.TunnelIP = s.allocIPLocked()
		}
	}
}

func (s *Server) edgeAllowedFor(e *model.Edge, orgID string) bool {
	if len(e.OrgIDs) == 0 {
		return true
	}
	for _, id := range e.OrgIDs {
		if id == orgID {
			return true
		}
	}
	return false
}