You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

356 lines
8.2 KiB

package cluster
import (
"fmt"
"io/fs"
"log"
"net/netip"
"strconv"
"git.jpi.io/amery/jpictl/pkg/rings"
"git.jpi.io/amery/jpictl/pkg/wireguard"
)
const (
// RingZeroPort is the port wireguard uses for ring0
RingZeroPort = 51800
// RingOnePort is the port wireguard uses for ring1
RingOnePort = 51810
)
// WireguardInterfaceID represents the number in the `wg%v`
// interface name.
type WireguardInterfaceID uint
// AsWireguardInterfaceID returns the [WireguardInterfaceID] for
// a valid [rings.RingID].
func AsWireguardInterfaceID(ring rings.RingID) (WireguardInterfaceID, error) {
switch ring {
case rings.RingZeroID:
return 0, nil
case rings.RingOneID:
return 1, nil
default:
return 0, ErrInvalidRing(ring)
}
}
// MustWireguardInterfaceID returns the [WireguardInterfaceID] for
// a valid [rings.RingID], and panics if it's not.
func MustWireguardInterfaceID(ring rings.RingID) WireguardInterfaceID {
id, err := AsWireguardInterfaceID(ring)
if err != nil {
panic(err)
}
return id
}
// RingID tells the [rings.RingID] of the [WireguardInterfaceID].
func (wi WireguardInterfaceID) RingID() rings.RingID {
return rings.RingID(wi + 1)
}
// PubFile returns "wgN.pub"
func (wi WireguardInterfaceID) PubFile() string {
return fmt.Sprintf("wg%v.pub", wi)
}
// KeyFile returns "wgN.key"
func (wi WireguardInterfaceID) KeyFile() string {
return fmt.Sprintf("wg%v.key", wi)
}
// ConfFile returns "wgN.conf"
func (wi WireguardInterfaceID) ConfFile() string {
return fmt.Sprintf("wg%v.conf", wi)
}
// Files returns all wgN.ext file names.
func (wi WireguardInterfaceID) Files() (keyFile, pubFile, confFile string) {
prefix := "wg" + strconv.Itoa(int(wi))
return prefix + ".key", prefix + ".pub", prefix + ".conf"
}
// RingInfo contains represents the Wireguard endpoint details
// for a Machine on a particular ring
type RingInfo struct {
Ring WireguardInterfaceID
Enabled bool
Keys wireguard.KeyPair
}
// RingID returns the [rings.RingID] for this [RingInfo].
func (ri *RingInfo) RingID() rings.RingID {
return rings.RingID(ri.Ring + 1)
}
// Merge attempts to combine two RingInfo structs
func (ri *RingInfo) Merge(alter *RingInfo) error {
switch {
case alter == nil:
return nil
case ri.Ring != alter.Ring:
// different ring
return fmt.Errorf("invalid %s: %v ≠ %v", "ring", ri.Ring, alter.Ring)
case ri.Enabled && !alter.Enabled:
// can't disable via Merge
return fmt.Errorf("invalid %s: %v → %v", "enabled", ri.Enabled, alter.Enabled)
case !canMergeKeyPairs(ri.Keys, alter.Keys):
// incompatible key pairs
return fmt.Errorf("invalid %s: %s ≠ %s", "keys", ri.Keys, alter.Keys)
}
return ri.unsafeMerge(alter)
}
func (ri *RingInfo) unsafeMerge(alter *RingInfo) error {
// enable via Merge
if alter.Enabled {
ri.Enabled = true
}
// fill the gaps on our key pair
if ri.Keys.PrivateKey.IsZero() {
ri.Keys.PrivateKey = alter.Keys.PrivateKey
}
if ri.Keys.PublicKey.IsZero() {
ri.Keys.PublicKey = alter.Keys.PublicKey
}
return nil
}
func canMergeKeyPairs(p1, p2 wireguard.KeyPair) bool {
switch {
case !p1.PrivateKey.IsZero() && !p2.PrivateKey.IsZero() && !p1.PrivateKey.Equal(p2.PrivateKey):
return false
case !p1.PublicKey.IsZero() && !p2.PublicKey.IsZero() && !p1.PublicKey.Equal(p2.PublicKey):
return false
default:
return true
}
}
// RingAddressEncoder provides encoder/decoder access for a particular
// Wireguard ring
type RingAddressEncoder struct {
ID rings.RingID
Port uint16
Encode func(rings.RegionID, rings.ZoneID, rings.NodeID) (netip.Addr, error)
Decode func(addr netip.Addr) (rings.RegionID, rings.ZoneID, rings.NodeID, bool)
}
var (
// RingZero is a wg0 address encoder/decoder
RingZero = RingAddressEncoder{
ID: rings.RingZeroID,
Port: RingZeroPort,
Decode: rings.DecodeRingZeroAddress,
Encode: rings.RingZeroAddress,
}
// RingOne is a wg1 address encoder/decoder
RingOne = RingAddressEncoder{
ID: rings.RingOneID,
Port: RingOnePort,
Decode: rings.DecodeRingOneAddress,
Encode: rings.RingOneAddress,
}
// Rings provides indexed access to the ring address encoders
Rings = []RingAddressEncoder{
RingZero,
RingOne,
}
)
var (
_ MachineIterator = (*Ring)(nil)
_ ZoneIterator = (*Ring)(nil)
)
// A Ring describes all peers on a ring
type Ring struct {
RingAddressEncoder
ZoneIterator
Peers []*RingPeer
}
// AddPeer adds a [Machine] to the ring
func (r *Ring) AddPeer(p *Machine) bool {
ri, ok := p.getRingInfo(r.ID)
if !ok {
return false
}
nodeID := p.ID
zoneID := p.Zone()
regionID := p.Region()
addr, _ := r.Encode(regionID, zoneID, nodeID)
rp := &RingPeer{
Node: p,
Address: addr,
PrivateKey: ri.Keys.PrivateKey,
PeerConfig: wireguard.PeerConfig{
Name: fmt.Sprintf("%s-%v", p.Name, ri.Ring),
PublicKey: ri.Keys.PublicKey,
Endpoint: wireguard.EndpointAddress{
Host: p.FullName(),
Port: r.Port,
},
},
}
switch {
case r.ID == rings.RingZeroID:
r.setRingZeroAllowedIPs(rp)
case p.IsGateway():
r.setRingOneGatewayAllowedIPs(rp)
default:
r.setRingOneNodeAllowedIPs(rp)
}
r.Peers = append(r.Peers, rp)
return true
}
func (r *Ring) setRingZeroAllowedIPs(rp *RingPeer) {
regionID, zoneID, _, _ := r.Decode(rp.Address)
// everyone on ring0 is a gateway to ring1
subnet, _ := rings.RingOnePrefix(regionID, zoneID)
rp.AllowSubnet(subnet)
// peer
rp.AllowCIDR(rp.Address, 32)
}
func (r *Ring) setRingOneGatewayAllowedIPs(rp *RingPeer) {
regionID, zoneID, _, _ := r.Decode(rp.Address)
// peer
rp.AllowCIDR(rp.Address, 32)
log.Println(rp.Node.Name, "0:", rp.Address, regionID, zoneID)
// ring1 gateways connect to all other ring1 networks
r.ForEachZone(func(z *Zone) bool {
log.Println(rp.Node.Name, "1:", z.Name, z.RegionID(), z.ID)
if !z.Is(regionID, zoneID) {
subnet := z.RingOnePrefix()
rp.AllowSubnet(subnet)
log.Println(rp.Node.Name, "1.1:", rp.PeerConfig.AllowedIPs)
}
return false
})
// ring1 gateways also connect to all ring0 addresses
r.ForEachZone(func(z *Zone) bool {
log.Println(rp.Node.Name, "2:", z.Name, z.RegionID(), z.ID)
z.ForEachMachine(func(p *Machine) bool {
log.Println(rp.Node.Name, "2.1:", p.Name, p.IsGateway())
if p.IsGateway() {
addr, _ := p.RingZeroAddress()
rp.AllowCIDR(addr, 32)
log.Println(rp.Node.Name, "2.2:", rp.PeerConfig.AllowedIPs)
}
return false
})
return false
})
log.Println(rp.Node.Name, "3:", rp.PeerConfig.AllowedIPs)
}
func (*Ring) setRingOneNodeAllowedIPs(rp *RingPeer) {
// only to the peer itself
rp.AllowCIDR(rp.Address, 32)
}
// ForEachMachine calls a function for each Machine in the ring
// until instructed to terminate the loop
func (r *Ring) ForEachMachine(fn func(*Machine) bool) {
for _, pp := range r.Peers {
if fn(pp.Node) {
return
}
}
}
// ExportConfig builds a wgN.conf for the specified machine on the ring
func (r *Ring) ExportConfig(p *Machine) (*wireguard.Config, error) {
var found bool
out := &wireguard.Config{
Interface: wireguard.InterfaceConfig{
ListenPort: r.Port,
},
}
for _, pp := range r.Peers {
switch {
case pp.Node == p:
// current
found = true
out.Interface.Name = pp.PeerConfig.Name
out.Interface.Address = pp.Address
out.Interface.PrivateKey = pp.PrivateKey
default:
// peer
pc := pp.PeerConfig
out.Peer = append(out.Peer, pc)
}
}
if !found {
return nil, fs.ErrNotExist
}
return out, nil
}
// A RingPeer is a node on a [Ring]
type RingPeer struct {
Node *Machine
Address netip.Addr
PrivateKey wireguard.PrivateKey
PeerConfig wireguard.PeerConfig
}
// AllowCIDR allows an IP range via this peer
func (rp *RingPeer) AllowCIDR(addr netip.Addr, bits int) {
rp.AllowSubnet(netip.PrefixFrom(addr, bits))
}
// AllowSubnet allows an IP range via this peer
func (rp *RingPeer) AllowSubnet(subnet netip.Prefix) {
rp.PeerConfig.AllowedIPs = append(rp.PeerConfig.AllowedIPs, subnet)
}
// NewRing composes a new Ring for Wireguard setup
func NewRing(z ZoneIterator, m MachineIterator, ringID rings.RingID) (*Ring, error) {
var r *Ring
for _, ring := range Rings {
if ringID == ring.ID {
r = &Ring{
RingAddressEncoder: ring,
ZoneIterator: z,
}
break
}
}
if r == nil {
return nil, ErrInvalidRing(ringID)
}
m.ForEachMachine(func(p *Machine) bool {
r.AddPeer(p)
return false
})
return r, nil
}