mesh.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819
  1. // Copyright 2019 the Kilo authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package mesh
  15. import (
  16. "bytes"
  17. "fmt"
  18. "io/ioutil"
  19. "net"
  20. "os"
  21. "sync"
  22. "time"
  23. "github.com/go-kit/kit/log"
  24. "github.com/go-kit/kit/log/level"
  25. "github.com/prometheus/client_golang/prometheus"
  26. "github.com/vishvananda/netlink"
  27. "github.com/squat/kilo/pkg/iproute"
  28. "github.com/squat/kilo/pkg/iptables"
  29. "github.com/squat/kilo/pkg/route"
  30. "github.com/squat/kilo/pkg/wireguard"
  31. )
  32. const resyncPeriod = 30 * time.Second
  33. const (
  34. // KiloPath is the directory where Kilo stores its configuration.
  35. KiloPath = "/var/lib/kilo"
  36. // PrivateKeyPath is the filepath where the WireGuard private key is stored.
  37. PrivateKeyPath = KiloPath + "/key"
  38. // ConfPath is the filepath where the WireGuard configuration is stored.
  39. ConfPath = KiloPath + "/conf"
  40. // DefaultKiloPort is the default UDP port Kilo uses.
  41. DefaultKiloPort = 51820
  42. // DefaultCNIPath is the default path to the CNI config file.
  43. DefaultCNIPath = "/etc/cni/net.d/10-kilo.conflist"
  44. )
  45. // DefaultKiloSubnet is the default CIDR for Kilo.
  46. var DefaultKiloSubnet = &net.IPNet{IP: []byte{10, 4, 0, 0}, Mask: []byte{255, 255, 0, 0}}
  47. // Granularity represents the abstraction level at which the network
  48. // should be meshed.
  49. type Granularity string
  50. // Encapsulate identifies what packets within a location should
  51. // be encapsulated.
  52. type Encapsulate string
  53. const (
  54. // LogicalGranularity indicates that the network should create
  55. // a mesh between logical locations, e.g. data-centers, but not between
  56. // all nodes within a single location.
  57. LogicalGranularity Granularity = "location"
  58. // FullGranularity indicates that the network should create
  59. // a mesh between every node.
  60. FullGranularity Granularity = "full"
  61. // NeverEncapsulate indicates that no packets within a location
  62. // should be encapsulated.
  63. NeverEncapsulate Encapsulate = "never"
  64. // CrossSubnetEncapsulate indicates that only packets that
  65. // traverse subnets within a location should be encapsulated.
  66. CrossSubnetEncapsulate Encapsulate = "crosssubnet"
  67. // AlwaysEncapsulate indicates that all packets within a location
  68. // should be encapsulated.
  69. AlwaysEncapsulate Encapsulate = "always"
  70. )
  71. // Node represents a node in the network.
  72. type Node struct {
  73. ExternalIP *net.IPNet
  74. Key []byte
  75. InternalIP *net.IPNet
  76. // LastSeen is a Unix time for the last time
  77. // the node confirmed it was live.
  78. LastSeen int64
  79. // Leader is a suggestion to Kilo that
  80. // the node wants to lead its segment.
  81. Leader bool
  82. Location string
  83. Name string
  84. Subnet *net.IPNet
  85. WireGuardIP *net.IPNet
  86. }
  87. // Ready indicates whether or not the node is ready.
  88. func (n *Node) Ready() bool {
  89. // Nodes that are not leaders will not have WireGuardIPs, so it is not required.
  90. return n != nil && n.ExternalIP != nil && n.Key != nil && n.InternalIP != nil && n.Subnet != nil && time.Now().Unix()-n.LastSeen < int64(resyncPeriod)*2/int64(time.Second)
  91. }
  92. // Peer represents a peer in the network.
  93. type Peer struct {
  94. wireguard.Peer
  95. Name string
  96. }
  97. // Ready indicates whether or not the peer is ready.
  98. func (p *Peer) Ready() bool {
  99. return p != nil && p.AllowedIPs != nil && len(p.AllowedIPs) != 0 && p.PublicKey != nil
  100. }
  101. // EventType describes what kind of an action an event represents.
  102. type EventType string
  103. const (
  104. // AddEvent represents an action where an item was added.
  105. AddEvent EventType = "add"
  106. // DeleteEvent represents an action where an item was removed.
  107. DeleteEvent EventType = "delete"
  108. // UpdateEvent represents an action where an item was updated.
  109. UpdateEvent EventType = "update"
  110. )
  111. // NodeEvent represents an event concerning a node in the cluster.
  112. type NodeEvent struct {
  113. Type EventType
  114. Node *Node
  115. Old *Node
  116. }
  117. // PeerEvent represents an event concerning a peer in the cluster.
  118. type PeerEvent struct {
  119. Type EventType
  120. Peer *Peer
  121. Old *Peer
  122. }
  123. // Backend can create clients for all of the
  124. // primitive types that Kilo deals with, namely:
  125. // * nodes; and
  126. // * peers.
  127. type Backend interface {
  128. Nodes() NodeBackend
  129. Peers() PeerBackend
  130. }
  131. // NodeBackend can get nodes by name, init itself,
  132. // list the nodes that should be meshed,
  133. // set Kilo properties for a node,
  134. // clean up any changes applied to the backend,
  135. // and watch for changes to nodes.
  136. type NodeBackend interface {
  137. CleanUp(string) error
  138. Get(string) (*Node, error)
  139. Init(<-chan struct{}) error
  140. List() ([]*Node, error)
  141. Set(string, *Node) error
  142. Watch() <-chan *NodeEvent
  143. }
  144. // PeerBackend can get peers by name, init itself,
  145. // list the peers that should be in the mesh,
  146. // set fields for a peer,
  147. // clean up any changes applied to the backend,
  148. // and watch for changes to peers.
  149. type PeerBackend interface {
  150. CleanUp(string) error
  151. Get(string) (*Peer, error)
  152. Init(<-chan struct{}) error
  153. List() ([]*Peer, error)
  154. Set(string, *Peer) error
  155. Watch() <-chan *PeerEvent
  156. }
  157. // Mesh is able to create Kilo network meshes.
  158. type Mesh struct {
  159. Backend
  160. cni bool
  161. cniPath string
  162. encapsulate Encapsulate
  163. externalIP *net.IPNet
  164. granularity Granularity
  165. hostname string
  166. internalIP *net.IPNet
  167. ipTables *iptables.Controller
  168. kiloIface int
  169. key []byte
  170. local bool
  171. port uint32
  172. priv []byte
  173. privIface int
  174. pub []byte
  175. pubIface int
  176. stop chan struct{}
  177. subnet *net.IPNet
  178. table *route.Table
  179. tunlIface int
  180. wireGuardIP *net.IPNet
  181. // nodes and peers are mutable fields in the struct
  182. // and needs to be guarded.
  183. nodes map[string]*Node
  184. peers map[string]*Peer
  185. mu sync.Mutex
  186. errorCounter *prometheus.CounterVec
  187. nodesGuage prometheus.Gauge
  188. peersGuage prometheus.Gauge
  189. reconcileCounter prometheus.Counter
  190. logger log.Logger
  191. }
  192. // New returns a new Mesh instance.
  193. func New(backend Backend, encapsulate Encapsulate, granularity Granularity, hostname string, port uint32, subnet *net.IPNet, local, cni bool, cniPath string, logger log.Logger) (*Mesh, error) {
  194. if err := os.MkdirAll(KiloPath, 0700); err != nil {
  195. return nil, fmt.Errorf("failed to create directory to store configuration: %v", err)
  196. }
  197. private, err := ioutil.ReadFile(PrivateKeyPath)
  198. private = bytes.Trim(private, "\n")
  199. if err != nil {
  200. level.Warn(logger).Log("msg", "no private key found on disk; generating one now")
  201. if private, err = wireguard.GenKey(); err != nil {
  202. return nil, err
  203. }
  204. }
  205. public, err := wireguard.PubKey(private)
  206. if err != nil {
  207. return nil, err
  208. }
  209. if err := ioutil.WriteFile(PrivateKeyPath, private, 0600); err != nil {
  210. return nil, fmt.Errorf("failed to write private key to disk: %v", err)
  211. }
  212. privateIP, publicIP, err := getIP(hostname)
  213. if err != nil {
  214. return nil, fmt.Errorf("failed to find public IP: %v", err)
  215. }
  216. ifaces, err := interfacesForIP(privateIP)
  217. if err != nil {
  218. return nil, fmt.Errorf("failed to find interface for private IP: %v", err)
  219. }
  220. privIface := ifaces[0].Index
  221. ifaces, err = interfacesForIP(publicIP)
  222. if err != nil {
  223. return nil, fmt.Errorf("failed to find interface for public IP: %v", err)
  224. }
  225. pubIface := ifaces[0].Index
  226. kiloIface, err := wireguard.New("kilo")
  227. if err != nil {
  228. return nil, fmt.Errorf("failed to create WireGuard interface: %v", err)
  229. }
  230. var tunlIface int
  231. if encapsulate != NeverEncapsulate {
  232. if tunlIface, err = iproute.NewIPIP(privIface); err != nil {
  233. return nil, fmt.Errorf("failed to create tunnel interface: %v", err)
  234. }
  235. if err := iproute.Set(tunlIface, true); err != nil {
  236. return nil, fmt.Errorf("failed to set tunnel interface up: %v", err)
  237. }
  238. }
  239. level.Debug(logger).Log("msg", fmt.Sprintf("using %s as the private IP address", privateIP.String()))
  240. level.Debug(logger).Log("msg", fmt.Sprintf("using %s as the public IP address", publicIP.String()))
  241. ipTables, err := iptables.New(len(subnet.IP))
  242. if err != nil {
  243. return nil, fmt.Errorf("failed to IP tables controller: %v", err)
  244. }
  245. return &Mesh{
  246. Backend: backend,
  247. cni: cni,
  248. cniPath: cniPath,
  249. encapsulate: encapsulate,
  250. externalIP: publicIP,
  251. granularity: granularity,
  252. hostname: hostname,
  253. internalIP: privateIP,
  254. ipTables: ipTables,
  255. kiloIface: kiloIface,
  256. nodes: make(map[string]*Node),
  257. peers: make(map[string]*Peer),
  258. port: port,
  259. priv: private,
  260. privIface: privIface,
  261. pub: public,
  262. pubIface: pubIface,
  263. local: local,
  264. stop: make(chan struct{}),
  265. subnet: subnet,
  266. table: route.NewTable(),
  267. tunlIface: tunlIface,
  268. errorCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
  269. Name: "kilo_errors_total",
  270. Help: "Number of errors that occurred while administering the mesh.",
  271. }, []string{"event"}),
  272. nodesGuage: prometheus.NewGauge(prometheus.GaugeOpts{
  273. Name: "kilo_nodes",
  274. Help: "Number of nodes in the mesh.",
  275. }),
  276. peersGuage: prometheus.NewGauge(prometheus.GaugeOpts{
  277. Name: "kilo_peers",
  278. Help: "Number of peers in the mesh.",
  279. }),
  280. reconcileCounter: prometheus.NewCounter(prometheus.CounterOpts{
  281. Name: "kilo_reconciles_total",
  282. Help: "Number of reconciliation attempts.",
  283. }),
  284. logger: logger,
  285. }, nil
  286. }
  287. // Run starts the mesh.
  288. func (m *Mesh) Run() error {
  289. if err := m.Nodes().Init(m.stop); err != nil {
  290. return fmt.Errorf("failed to initialize node backend: %v", err)
  291. }
  292. if err := m.Peers().Init(m.stop); err != nil {
  293. return fmt.Errorf("failed to initialize peer backend: %v", err)
  294. }
  295. ipTablesErrors, err := m.ipTables.Run(m.stop)
  296. if err != nil {
  297. return fmt.Errorf("failed to watch for IP tables updates: %v", err)
  298. }
  299. routeErrors, err := m.table.Run(m.stop)
  300. if err != nil {
  301. return fmt.Errorf("failed to watch for route table updates: %v", err)
  302. }
  303. go func() {
  304. for {
  305. var err error
  306. select {
  307. case err = <-ipTablesErrors:
  308. case err = <-routeErrors:
  309. case <-m.stop:
  310. return
  311. }
  312. if err != nil {
  313. level.Error(m.logger).Log("error", err)
  314. m.errorCounter.WithLabelValues("run").Inc()
  315. }
  316. }
  317. }()
  318. defer m.cleanUp()
  319. t := time.NewTimer(resyncPeriod)
  320. nw := m.Nodes().Watch()
  321. pw := m.Peers().Watch()
  322. var ne *NodeEvent
  323. var pe *PeerEvent
  324. for {
  325. select {
  326. case ne = <-nw:
  327. m.syncNodes(ne)
  328. case pe = <-pw:
  329. m.syncPeers(pe)
  330. case <-t.C:
  331. m.checkIn()
  332. if m.cni {
  333. m.updateCNIConfig()
  334. }
  335. m.syncEndpoints()
  336. m.applyTopology()
  337. t.Reset(resyncPeriod)
  338. case <-m.stop:
  339. return nil
  340. }
  341. }
  342. }
  343. // WireGuard updates the endpoints of peers to match the
  344. // last place a valid packet was received from.
  345. // Periodically we need to syncronize the endpoints
  346. // of peers in the backend to match the WireGuard configuration.
  347. func (m *Mesh) syncEndpoints() {
  348. link, err := linkByIndex(m.kiloIface)
  349. if err != nil {
  350. level.Error(m.logger).Log("error", err)
  351. m.errorCounter.WithLabelValues("endpoints").Inc()
  352. return
  353. }
  354. conf, err := wireguard.ShowConf(link.Attrs().Name)
  355. if err != nil {
  356. level.Error(m.logger).Log("error", err)
  357. m.errorCounter.WithLabelValues("endpoints").Inc()
  358. return
  359. }
  360. m.mu.Lock()
  361. defer m.mu.Unlock()
  362. c := wireguard.Parse(conf)
  363. var key string
  364. var tmp *Peer
  365. for i := range c.Peers {
  366. // Peers are indexed by public key.
  367. key = string(c.Peers[i].PublicKey)
  368. if p, ok := m.peers[key]; ok {
  369. tmp = &Peer{
  370. Name: p.Name,
  371. Peer: *c.Peers[i],
  372. }
  373. if !peersAreEqual(tmp, p) {
  374. p.Endpoint = tmp.Endpoint
  375. if err := m.Peers().Set(p.Name, p); err != nil {
  376. level.Error(m.logger).Log("error", err)
  377. m.errorCounter.WithLabelValues("endpoints").Inc()
  378. }
  379. }
  380. }
  381. }
  382. }
  383. func (m *Mesh) syncNodes(e *NodeEvent) {
  384. logger := log.With(m.logger, "event", e.Type)
  385. level.Debug(logger).Log("msg", "syncing nodes", "event", e.Type)
  386. if isSelf(m.hostname, e.Node) {
  387. level.Debug(logger).Log("msg", "processing local node", "node", e.Node)
  388. m.handleLocal(e.Node)
  389. return
  390. }
  391. var diff bool
  392. m.mu.Lock()
  393. if !e.Node.Ready() {
  394. level.Debug(logger).Log("msg", "received incomplete node", "node", e.Node)
  395. // An existing node is no longer valid
  396. // so remove it from the mesh.
  397. if _, ok := m.nodes[e.Node.Name]; ok {
  398. level.Info(logger).Log("msg", "node is no longer ready", "node", e.Node)
  399. diff = true
  400. }
  401. } else {
  402. switch e.Type {
  403. case AddEvent:
  404. fallthrough
  405. case UpdateEvent:
  406. if !nodesAreEqual(m.nodes[e.Node.Name], e.Node) {
  407. diff = true
  408. }
  409. // Even if the nodes are the same,
  410. // overwrite the old node to update the timestamp.
  411. m.nodes[e.Node.Name] = e.Node
  412. case DeleteEvent:
  413. delete(m.nodes, e.Node.Name)
  414. diff = true
  415. }
  416. }
  417. m.mu.Unlock()
  418. if diff {
  419. level.Info(logger).Log("node", e.Node)
  420. m.applyTopology()
  421. }
  422. }
  423. func (m *Mesh) syncPeers(e *PeerEvent) {
  424. logger := log.With(m.logger, "event", e.Type)
  425. level.Debug(logger).Log("msg", "syncing peers", "event", e.Type)
  426. var diff bool
  427. m.mu.Lock()
  428. // Peers are indexed by public key.
  429. key := string(e.Peer.PublicKey)
  430. if !e.Peer.Ready() {
  431. level.Debug(logger).Log("msg", "received incomplete peer", "peer", e.Peer)
  432. // An existing peer is no longer valid
  433. // so remove it from the mesh.
  434. if _, ok := m.peers[key]; ok {
  435. level.Info(logger).Log("msg", "peer is no longer ready", "peer", e.Peer)
  436. diff = true
  437. }
  438. } else {
  439. switch e.Type {
  440. case AddEvent:
  441. fallthrough
  442. case UpdateEvent:
  443. if e.Old != nil && key != string(e.Old.PublicKey) {
  444. delete(m.peers, string(e.Old.PublicKey))
  445. diff = true
  446. }
  447. if !peersAreEqual(m.peers[key], e.Peer) {
  448. m.peers[key] = e.Peer
  449. diff = true
  450. }
  451. case DeleteEvent:
  452. delete(m.peers, key)
  453. diff = true
  454. }
  455. }
  456. m.mu.Unlock()
  457. if diff {
  458. level.Info(logger).Log("peer", e.Peer)
  459. m.applyTopology()
  460. }
  461. }
  462. // checkIn will try to update the local node's LastSeen timestamp
  463. // in the backend.
  464. func (m *Mesh) checkIn() {
  465. m.mu.Lock()
  466. defer m.mu.Unlock()
  467. n := m.nodes[m.hostname]
  468. if n == nil {
  469. level.Debug(m.logger).Log("msg", "no local node found in backend")
  470. return
  471. }
  472. oldTime := n.LastSeen
  473. n.LastSeen = time.Now().Unix()
  474. if err := m.Nodes().Set(m.hostname, n); err != nil {
  475. level.Error(m.logger).Log("error", fmt.Sprintf("failed to set local node: %v", err), "node", n)
  476. m.errorCounter.WithLabelValues("checkin").Inc()
  477. // Revert time.
  478. n.LastSeen = oldTime
  479. return
  480. }
  481. level.Debug(m.logger).Log("msg", "successfully checked in local node in backend")
  482. }
  483. func (m *Mesh) handleLocal(n *Node) {
  484. // Allow the external IP to be overridden.
  485. if n.ExternalIP == nil {
  486. n.ExternalIP = m.externalIP
  487. }
  488. // Compare the given node to the calculated local node.
  489. // Take leader, location, and subnet from the argument, as these
  490. // are not determined by kilo.
  491. local := &Node{
  492. ExternalIP: n.ExternalIP,
  493. Key: m.pub,
  494. InternalIP: m.internalIP,
  495. LastSeen: time.Now().Unix(),
  496. Leader: n.Leader,
  497. Location: n.Location,
  498. Name: m.hostname,
  499. Subnet: n.Subnet,
  500. WireGuardIP: m.wireGuardIP,
  501. }
  502. if !nodesAreEqual(n, local) {
  503. level.Debug(m.logger).Log("msg", "local node differs from backend")
  504. if err := m.Nodes().Set(m.hostname, local); err != nil {
  505. level.Error(m.logger).Log("error", fmt.Sprintf("failed to set local node: %v", err), "node", local)
  506. m.errorCounter.WithLabelValues("local").Inc()
  507. return
  508. }
  509. level.Debug(m.logger).Log("msg", "successfully reconciled local node against backend")
  510. }
  511. m.mu.Lock()
  512. n = m.nodes[m.hostname]
  513. if n == nil {
  514. n = &Node{}
  515. }
  516. m.mu.Unlock()
  517. if !nodesAreEqual(n, local) {
  518. m.mu.Lock()
  519. m.nodes[local.Name] = local
  520. m.mu.Unlock()
  521. m.applyTopology()
  522. }
  523. }
  524. func (m *Mesh) applyTopology() {
  525. m.reconcileCounter.Inc()
  526. m.mu.Lock()
  527. defer m.mu.Unlock()
  528. // Ensure only ready nodes are considered.
  529. nodes := make(map[string]*Node)
  530. var readyNodes float64
  531. for k := range m.nodes {
  532. if !m.nodes[k].Ready() {
  533. continue
  534. }
  535. nodes[k] = m.nodes[k]
  536. readyNodes++
  537. }
  538. // Ensure only ready nodes are considered.
  539. peers := make(map[string]*Peer)
  540. var readyPeers float64
  541. for k := range m.peers {
  542. if !m.peers[k].Ready() {
  543. continue
  544. }
  545. peers[k] = m.peers[k]
  546. readyPeers++
  547. }
  548. m.nodesGuage.Set(readyNodes)
  549. m.peersGuage.Set(readyPeers)
  550. // We cannot do anything with the topology until the local node is available.
  551. if nodes[m.hostname] == nil {
  552. return
  553. }
  554. t, err := NewTopology(nodes, peers, m.granularity, m.hostname, m.port, m.priv, m.subnet)
  555. if err != nil {
  556. level.Error(m.logger).Log("error", err)
  557. m.errorCounter.WithLabelValues("apply").Inc()
  558. return
  559. }
  560. // Update the node's WireGuard IP.
  561. m.wireGuardIP = t.wireGuardCIDR
  562. conf := t.Conf()
  563. buf, err := conf.Bytes()
  564. if err != nil {
  565. level.Error(m.logger).Log("error", err)
  566. m.errorCounter.WithLabelValues("apply").Inc()
  567. }
  568. if err := ioutil.WriteFile(ConfPath, buf, 0600); err != nil {
  569. level.Error(m.logger).Log("error", err)
  570. m.errorCounter.WithLabelValues("apply").Inc()
  571. return
  572. }
  573. rules := iptables.ForwardRules(m.subnet)
  574. var peerCIDRs []*net.IPNet
  575. for _, p := range peers {
  576. rules = append(rules, iptables.ForwardRules(p.AllowedIPs...)...)
  577. peerCIDRs = append(peerCIDRs, p.AllowedIPs...)
  578. }
  579. rules = append(rules, iptables.MasqueradeRules(m.subnet, oneAddressCIDR(t.privateIP.IP), nodes[m.hostname].Subnet, t.RemoteSubnets(), peerCIDRs)...)
  580. // If we are handling local routes, ensure the local
  581. // tunnel has an IP address and IPIP traffic is allowed.
  582. if m.encapsulate != NeverEncapsulate && m.local {
  583. var cidrs []*net.IPNet
  584. for _, s := range t.segments {
  585. if s.location == nodes[m.hostname].Location {
  586. for i := range s.privateIPs {
  587. cidrs = append(cidrs, oneAddressCIDR(s.privateIPs[i]))
  588. }
  589. break
  590. }
  591. }
  592. rules = append(rules, iptables.EncapsulateRules(cidrs)...)
  593. // If we are handling local routes, ensure the local
  594. // tunnel has an IP address.
  595. if err := iproute.SetAddress(m.tunlIface, oneAddressCIDR(newAllocator(*nodes[m.hostname].Subnet).next().IP)); err != nil {
  596. level.Error(m.logger).Log("error", err)
  597. m.errorCounter.WithLabelValues("apply").Inc()
  598. return
  599. }
  600. }
  601. if err := m.ipTables.Set(rules); err != nil {
  602. level.Error(m.logger).Log("error", err)
  603. m.errorCounter.WithLabelValues("apply").Inc()
  604. return
  605. }
  606. if t.leader {
  607. if err := iproute.SetAddress(m.kiloIface, t.wireGuardCIDR); err != nil {
  608. level.Error(m.logger).Log("error", err)
  609. m.errorCounter.WithLabelValues("apply").Inc()
  610. return
  611. }
  612. link, err := linkByIndex(m.kiloIface)
  613. if err != nil {
  614. level.Error(m.logger).Log("error", err)
  615. m.errorCounter.WithLabelValues("apply").Inc()
  616. return
  617. }
  618. oldConf, err := wireguard.ShowConf(link.Attrs().Name)
  619. if err != nil {
  620. level.Error(m.logger).Log("error", err)
  621. m.errorCounter.WithLabelValues("apply").Inc()
  622. return
  623. }
  624. // Setting the WireGuard configuration interrupts existing connections
  625. // so only set the configuration if it has changed.
  626. equal := conf.Equal(wireguard.Parse(oldConf))
  627. if !equal {
  628. level.Info(m.logger).Log("msg", "WireGuard configurations are different")
  629. if err := wireguard.SetConf(link.Attrs().Name, ConfPath); err != nil {
  630. level.Error(m.logger).Log("error", err)
  631. m.errorCounter.WithLabelValues("apply").Inc()
  632. return
  633. }
  634. }
  635. if err := iproute.Set(m.kiloIface, true); err != nil {
  636. level.Error(m.logger).Log("error", err)
  637. m.errorCounter.WithLabelValues("apply").Inc()
  638. return
  639. }
  640. } else {
  641. level.Debug(m.logger).Log("msg", "local node is not the leader")
  642. if err := iproute.Set(m.kiloIface, false); err != nil {
  643. level.Error(m.logger).Log("error", err)
  644. m.errorCounter.WithLabelValues("apply").Inc()
  645. return
  646. }
  647. }
  648. // We need to add routes last since they may depend
  649. // on the WireGuard interface.
  650. routes := t.Routes(m.kiloIface, m.privIface, m.tunlIface, m.local, m.encapsulate)
  651. if err := m.table.Set(routes); err != nil {
  652. level.Error(m.logger).Log("error", err)
  653. m.errorCounter.WithLabelValues("apply").Inc()
  654. }
  655. }
  656. // RegisterMetrics registers Prometheus metrics on the given Prometheus
  657. // registerer.
  658. func (m *Mesh) RegisterMetrics(r prometheus.Registerer) {
  659. r.MustRegister(
  660. m.errorCounter,
  661. m.nodesGuage,
  662. m.peersGuage,
  663. m.reconcileCounter,
  664. )
  665. }
  666. // Stop stops the mesh.
  667. func (m *Mesh) Stop() {
  668. close(m.stop)
  669. }
  670. func (m *Mesh) cleanUp() {
  671. if err := m.ipTables.CleanUp(); err != nil {
  672. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up IP tables: %v", err))
  673. m.errorCounter.WithLabelValues("cleanUp").Inc()
  674. }
  675. if err := m.table.CleanUp(); err != nil {
  676. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up routes: %v", err))
  677. m.errorCounter.WithLabelValues("cleanUp").Inc()
  678. }
  679. if err := os.Remove(PrivateKeyPath); err != nil {
  680. level.Error(m.logger).Log("error", fmt.Sprintf("failed to delete private key: %v", err))
  681. m.errorCounter.WithLabelValues("cleanUp").Inc()
  682. }
  683. if err := os.Remove(ConfPath); err != nil {
  684. level.Error(m.logger).Log("error", fmt.Sprintf("failed to delete configuration file: %v", err))
  685. m.errorCounter.WithLabelValues("cleanUp").Inc()
  686. }
  687. if err := iproute.RemoveInterface(m.kiloIface); err != nil {
  688. level.Error(m.logger).Log("error", fmt.Sprintf("failed to remove WireGuard interface: %v", err))
  689. m.errorCounter.WithLabelValues("cleanUp").Inc()
  690. }
  691. if err := m.Nodes().CleanUp(m.hostname); err != nil {
  692. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up node backend: %v", err))
  693. m.errorCounter.WithLabelValues("cleanUp").Inc()
  694. }
  695. if err := m.Peers().CleanUp(m.hostname); err != nil {
  696. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up peer backend: %v", err))
  697. m.errorCounter.WithLabelValues("cleanUp").Inc()
  698. }
  699. }
  700. func isSelf(hostname string, node *Node) bool {
  701. return node != nil && node.Name == hostname
  702. }
  703. func nodesAreEqual(a, b *Node) bool {
  704. if !(a != nil) == (b != nil) {
  705. return false
  706. }
  707. if a == b {
  708. return true
  709. }
  710. // Ignore LastSeen when comparing equality we want to check if the nodes are
  711. // equivalent. However, we do want to check if LastSeen has transitioned
  712. // between valid and invalid.
  713. return ipNetsEqual(a.ExternalIP, b.ExternalIP) && string(a.Key) == string(b.Key) && ipNetsEqual(a.WireGuardIP, b.WireGuardIP) && ipNetsEqual(a.InternalIP, b.InternalIP) && a.Leader == b.Leader && a.Location == b.Location && a.Name == b.Name && subnetsEqual(a.Subnet, b.Subnet) && a.Ready() == b.Ready()
  714. }
  715. func peersAreEqual(a, b *Peer) bool {
  716. if !(a != nil) == (b != nil) {
  717. return false
  718. }
  719. if a == b {
  720. return true
  721. }
  722. if !(a.Endpoint != nil) == (b.Endpoint != nil) {
  723. return false
  724. }
  725. if a.Endpoint != nil {
  726. if !a.Endpoint.IP.Equal(b.Endpoint.IP) || a.Endpoint.Port != b.Endpoint.Port {
  727. return false
  728. }
  729. }
  730. if len(a.AllowedIPs) != len(b.AllowedIPs) {
  731. return false
  732. }
  733. for i := range a.AllowedIPs {
  734. if !ipNetsEqual(a.AllowedIPs[i], b.AllowedIPs[i]) {
  735. return false
  736. }
  737. }
  738. return string(a.PublicKey) == string(b.PublicKey) && a.PersistentKeepalive == b.PersistentKeepalive
  739. }
  740. func ipNetsEqual(a, b *net.IPNet) bool {
  741. if a == nil && b == nil {
  742. return true
  743. }
  744. if (a != nil) != (b != nil) {
  745. return false
  746. }
  747. if a.Mask.String() != b.Mask.String() {
  748. return false
  749. }
  750. return a.IP.Equal(b.IP)
  751. }
  752. func subnetsEqual(a, b *net.IPNet) bool {
  753. if a == nil && b == nil {
  754. return true
  755. }
  756. if (a != nil) != (b != nil) {
  757. return false
  758. }
  759. if a.Mask.String() != b.Mask.String() {
  760. return false
  761. }
  762. if !a.Contains(b.IP) {
  763. return false
  764. }
  765. if !b.Contains(a.IP) {
  766. return false
  767. }
  768. return true
  769. }
  770. func linkByIndex(index int) (netlink.Link, error) {
  771. link, err := netlink.LinkByIndex(index)
  772. if err != nil {
  773. return nil, fmt.Errorf("failed to get interface: %v", err)
  774. }
  775. return link, nil
  776. }