srv support

Author: nyiyui

api/main.go

@@ -20,6 +20,20 @@ type PushS struct {
 	PubKeys map[string]wgtypes.Key
 }
 
+type SRVUpdateQ struct {
+	CentralToken util.Token
+	SRVs         []SRV
+}
+
+type SRVUpdateS struct{}
+
+type SRV struct {
+	NetworkName string
+	PeerName    string
+
+	central.SRV
+}
+
 type GenerateQ struct {
 	CNNs []string
 }

central/main.go

@@ -3,6 +3,7 @@ package central
 
 import (
 	"encoding/json"
+	"errors"
 	"net"
 	"time"
 
@@ -53,10 +54,12 @@ type Peer struct {
 	CanForward bool    `yaml:"canForward" json:"canForward"`
 	// CanSee determines whether this Peer can see anything (nil) or specfic peers only (non-nil).
 	// TODO: when CanSee.Only is blank, this is interpreted as nil → no way to distinguish between seeing nothing and everything
-	CanSee *CanSee `yaml:"canSee" json:"canSee"`
+	CanSee      *CanSee `yaml:"canSee" json:"canSee"`
+	AllowedSRVs []SRVAllowance
 
 	PubKey          wgtypes.Key
 	ForwardingPeers []string
+	SRVs            []SRV
 }
 
 func (p *Peer) String() string {
@@ -86,6 +89,73 @@ func (c *CanSee) Same(c2 *CanSee) bool {
 	return Same3(c.Only, c2.Only)
 }
 
+type SRVAllowable interface {
+	// AllowedBy returns nil this is allowed by the SRVAllowance, and a non-nil error otherwise.
+	AllowedBy(SRVAllowance) error
+}
+
+func AllowedByAny(sa SRVAllowable, a2s []SRVAllowance) bool {
+	for _, a2 := range a2s {
+		err := sa.AllowedBy(a2)
+		if err == nil {
+			return true
+		}
+	}
+	return false
+}
+
+type SRVAllowance struct {
+	Service     string
+	ServiceAny  bool
+	Name        string
+	NameAny     bool
+	PriorityMin uint16
+	PriorityMax uint16
+	WeightMin   uint16
+	WeightMax   uint16
+}
+
+func (a SRVAllowance) AllowedBy(a2 SRVAllowance) error {
+	if !a2.ServiceAny && a.Service != a2.Service {
+		return errors.New("Service mismatch")
+	}
+	if !a2.NameAny && a.Name != a2.Name {
+		return errors.New("Name mismatch")
+	}
+	if !(a.PriorityMin >= a2.PriorityMin && a.PriorityMax <= a2.PriorityMax) {
+		return errors.New("Priority not in range")
+	}
+	if !(a.WeightMin >= a2.WeightMin && a.WeightMax <= a2.WeightMax) {
+		return errors.New("Weight not in range")
+	}
+	return nil
+}
+
+type SRV struct {
+	Service  string
+	Protocol string
+	Name     string
+	Priority uint16
+	Weight   uint16
+	Port     uint16
+}
+
+func (s SRV) AllowedBy(a2 SRVAllowance) error {
+	if !a2.ServiceAny && s.Service != a2.Service {
+		return errors.New("Service mismatch")
+	}
+	if !a2.NameAny && s.Name != a2.Name {
+		return errors.New("Name mismatch")
+	}
+	if !(a2.PriorityMin <= s.Priority && s.Priority <= a2.PriorityMax) {
+		return errors.New("Priority not in range")
+	}
+	if !(a2.WeightMin <= s.Weight && s.Weight <= a2.WeightMax) {
+		return errors.New("Weight not in range")
+	}
+	return nil
+}
+
 // Duration is a encoding-friendly time.Duration.
 type Duration time.Duration
 

cmd/runner-node/main.go

@@ -23,6 +23,7 @@ type config struct {
 	CS               csConfig     `yaml:"cs"`
 	Hokuto           hokutoConfig `yaml:"hokuto"`
 	EndpointOverride string       `yaml:"endpointOverride"`
+	SRVList          string       `yaml:"srvList"`
 }
 
 type hokutoConfig struct {
@@ -130,6 +131,7 @@ func main() {
 		HokutoExtraParents: c.Hokuto.ExtraParents,
 		EndpointOverride:   c.EndpointOverride,
 		BackportPath:       filepath.Join(os.Getenv("STATE_DIRECTORY"), "node-backport.json"),
+		SRVList:            c.SRVList,
 	}
 	if os.Getenv("HOKUTO_ADDR") != "" {
 		nc.HokutoAddr = os.Getenv("HOKUTO_ADDR")

cs/api.go

@@ -16,6 +16,7 @@ func (c *CentralSource) newHandler() {
 	h.Handle("ping", c.ping)
 	h.Handle("sync", c.sync)
 	h.Handle("azusa", c.azusa)
+	h.Handle("srvUpdate", c.srvUpdate)
 	c.handler = h
 }
 

cs/azusa.go

@@ -26,17 +26,25 @@ func (c *CentralSource) azusa(cl *rpc2.Client, q *api.AzusaQ, s *api.AzusaS) err
 		}
 		peer, ok := cn.Peers[q.Networks[cnn].Name]
 		if !ok {
-			return fmt.Errorf("net %s no exist :(", cnn)
+			return fmt.Errorf("net %s peer %s no exist :(", cnn, q.Networks[cnn].Name)
 		}
 		peer.Name = q.Networks[cnn].Name
-		err = checkPeer(ti, cnn, peer)
+		err = checkPeer(ti, cnn, *peer)
 		if err != nil {
 			return err
 		}
-		_, ok := c.cc.Networks[cnn]
+		_, ok = c.cc.Networks[cnn]
 		if !ok {
 			return fmt.Errorf("net %s no exist :(", cnn)
 		}
+		if !ti.SRVAllowancesAny {
+			for saI, sa := range q.Networks[cnn].AllowedSRVs {
+				if !central.AllowedByAny(sa, ti.SRVAllowances) {
+					return fmt.Errorf("peer allowance %d: not allowed by any token-level allowances", saI)
+				}
+			}
+		}
+		// TODO: token-level restrictions on SRVAllowance and SRVs
 		fmt.Fprintf(&desc, "\n- net %s peer %s: %#v", cnn, peer.Name, peer)
 	}
 	util.S.Infof("azusa from token %s to push %d:\n%s", ti.Name, len(q.Networks), &desc)

cs/srvupdate.go

@@ -0,0 +1,73 @@
+package cs
+
+import (
+	"fmt"
+
+	"github.com/cenkalti/rpc2"
+	"github.com/nyiyui/qrystal/api"
+	"github.com/nyiyui/qrystal/central"
+	"github.com/nyiyui/qrystal/util"
+	"golang.org/x/exp/slices"
+)
+
+func (c *CentralSource) srvUpdate(cl *rpc2.Client, q *api.SRVUpdateQ, s *api.SRVUpdateS) error {
+	ti, ok, err := c.Tokens.getToken(&q.CentralToken)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return newTokenAuthError(q.CentralToken)
+	}
+	for _, srv := range q.SRVs {
+		cn, ok := c.cc.Networks[srv.NetworkName]
+		if !ok {
+			return fmt.Errorf("net %s no exist :(", srv.NetworkName)
+		}
+		peer, ok := cn.Peers[srv.PeerName]
+		if !ok {
+			return fmt.Errorf("net %s peer %s no exist :(", srv.NetworkName, srv.PeerName)
+		}
+		err = checkPeer(ti, srv.NetworkName, *peer)
+		if err != nil {
+			return err
+		}
+	}
+	util.S.Infof("srvUpdate from token %s to push %d:\n%#v", ti.Name, len(q.SRVs), q.SRVs)
+	ti.StartUse()
+	err = c.Tokens.UpdateToken(ti)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		ti.StopUse()
+		err = c.Tokens.UpdateToken(ti)
+		if err != nil {
+			util.S.Errorf("UpdateToken %s: %s", ti.key, err)
+		}
+	}()
+	c.ccLock.Lock()
+	defer c.ccLock.Unlock()
+	for srvI, srv := range q.SRVs {
+		cn, ok := c.cc.Networks[srv.NetworkName]
+		if !ok {
+			return fmt.Errorf("net %s no exist :(", srv.NetworkName)
+		}
+		peer, ok := cn.Peers[srv.PeerName]
+		if !ok {
+			return fmt.Errorf("net %s peer %s no exist :(", srv.NetworkName, srv.PeerName)
+		}
+		if !central.AllowedByAny(srv, peer.AllowedSRVs) {
+			return fmt.Errorf("srv %d: not allowed", srvI)
+		}
+		i := slices.IndexFunc(peer.SRVs, func(s central.SRV) bool { return s.Service == srv.Service })
+		if srv.Name == "" && i != -1 {
+			peer.SRVs = append(peer.SRVs[:i], peer.SRVs[i+1:]...)
+		} else if i == -1 {
+			i = len(peer.SRVs)
+			peer.SRVs = append(peer.SRVs, srv.SRV)
+		} else {
+			peer.SRVs[i] = srv.SRV
+		}
+	}
+	return nil
+}

cs/token.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/nyiyui/qrystal/central"
 	"github.com/nyiyui/qrystal/util"
 	"github.com/tidwall/buntdb"
 	"gopkg.in/yaml.v3"
@@ -126,13 +127,19 @@ func (s *TokenStore) convertToMap() (m map[string]string, err error) {
 }
 
 type TokenInfo struct {
-	key            string `json:"-"`
-	Name           string
-	Networks       map[string]string
-	CanPull        bool
-	CanPush        *CanPush
-	CanAdminTokens *CanAdminTokens
+	key      string `json:"-"`
+	Name     string
+	Networks map[string]string
+	CanPull  bool
+	CanPush  *CanPush
 	// CanAdminTokens specifies whether this token can add *or remove* tokens.
+	CanAdminTokens *CanAdminTokens
+	// CanSRVUpdate allows updating SRV allowances for peers it can push to.
+	CanSRVUpdate bool
+	// SRVAllowance is a ORed list of SRV alloances for peers it can push to.
+	SRVAllowances []central.SRVAllowance
+	// SRVAllowancesAny allows updating SRV allowances without restrictions by SRVAllowances.
+	SRVAllowancesAny bool
 
 	Using    bool
 	LastUsed time.Time

node/loop.go

@@ -4,6 +4,9 @@ package node
 
 import (
 	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"github.com/cenkalti/rpc2"
@@ -17,15 +20,48 @@ type listenError struct {
 }
 
 func (n *Node) ListenCS() {
+	go func() {
+		err := n.handleSRV()
+		util.S.Errorf("srv error: %s", err)
+		panic(fmt.Sprintf("srv error: %s", err))
+	}()
 	util.S.Debug("listening…")
 	err := n.listenCS()
 	util.S.Errorf("cs listen error: %s", err)
 }
 
+func (n *Node) handleSRV() error {
+	reloadCh := make(chan os.Signal)
+	signal.Notify(reloadCh, syscall.SIGUSR1)
+	return util.Backoff(func() (resetBackoff bool, err error) { return n.handleSRVOnce(reloadCh) }, func(backoff time.Duration, err error) error {
+		util.S.Errorf("srv: %s; retry in %s", err, backoff)
+		util.S.Errorw("srv: error",
+			"err", err,
+			"backoff", backoff,
+		)
+		return nil
+	})
+}
+
+func (n *Node) handleSRVOnce(reloadCh <-chan os.Signal) (resetBackoff bool, err error) {
+	util.S.Debug("newClient…")
+	cl, _, err := n.newClient()
+	if err != nil {
+		return false, fmt.Errorf("newClient: %w", err)
+	}
+
+	for range reloadCh {
+		err = n.loadSRVList(cl)
+		if err != nil {
+			err = fmt.Errorf("srv: %w", err)
+			return
+		}
+	}
+	return true, nil
+}
+
 func (n *Node) listenCS() error {
-	return util.Backoff(func() (resetBackoff bool, err error) {
-		return n.listenCSOnce()
-	}, func(backoff time.Duration, err error) error {
+	return util.Backoff(n.listenCSOnce, func(backoff time.Duration, err error) error {
 		util.Notify(fmt.Sprintf("STATUS=connecting to CS: %s (retrying in %s)", err, backoff))
 		util.S.Errorf("listen: %s; retry in %s", err, backoff)
 		util.S.Errorw("listen: error",
@@ -37,7 +73,6 @@ func (n *Node) listenCS() error {
 }
 
 func (n *Node) listenCSOnce() (resetBackoff bool, err error) {
-	// Setup
 	util.S.Debug("newClient…")
 	cl, _, err := n.newClient()
 	if err != nil {

node/main.go

@@ -23,6 +23,7 @@ type NodeConfig struct {
 	CS                 CSConfig
 	EndpointOverride   string
 	BackportPath       string
+	SRVList            string
 }
 
 // There must be only one Node instance as a Node can trigger a trace to stop.
@@ -48,6 +49,7 @@ func NewNode(cfg NodeConfig) (*Node, error) {
 		hokuto:               hh,
 		endpointOverridePath: cfg.EndpointOverride,
 		backportPath:         cfg.BackportPath,
+		srvListPath:          cfg.SRVList,
 	}
 	if cfg.HokutoDNSAddr != "" {
 		addr, err := net.ResolveUDPAddr("udp", cfg.HokutoDNSAddr)
@@ -84,4 +86,5 @@ type Node struct {
 	eoStateLock          sync.Mutex
 
 	backportPath string
+	srvListPath  string
 }