zishuo/netmaker
1
0
Fork 0
mirror of https://github.com/gravitl/netmaker.git synced 2025-10-10 03:00:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

added relay functionality to proxy

Browse Source
This commit is contained in:
Abhishek Kondur
2022-11-04 01:24:48 +05:30
parent 4e55242cb1
commit acae6c3aed
17 changed files with 430 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
controllers/relay.go
View File

@@ -10,6 +10,8 @@ import (
"github.com/gravitl/netmaker/logic"
"github.com/gravitl/netmaker/models"
"github.com/gravitl/netmaker/mq"
"github.com/gravitl/netmaker/nm-proxy/manager"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
)
// swagger:route POST /api/nodes/{network}/{nodeid}/createrelay nodes createRelay
@@ -42,13 +44,50 @@ func createRelay(w http.ResponseWriter, r *http.Request) {
logic.ReturnErrorResponse(w, r, logic.FormatError(err, "internal"))
return
}
relayPeersMap := make(map[string][]wgtypes.PeerConfig)
logger.Log(1, r.Header.Get("user"), "created relay on node", relay.NodeID, "on network", relay.NetID)
for _, relayedNode := range updatenodes {
peers, err := logic.GetPeersForProxy(&relayedNode)
if err == nil {
relayPeersMap[relayedNode.PublicKey] = peers
}
// relayEndpoint, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", node.Endpoint, node.LocalListenPort))
// if err != nil {
// logger.Log(1, "failed to resolve relay node endpoint: ", err.Error())
// }
// err = mq.ProxyUpdate(&manager.ManagerAction{
// Action: manager.AddInterface,
// Payload: manager.ManagerPayload{
// InterfaceName: relayedNode.Interface,
// IsRelayed: true,
// Peers: peers,
// RelayedTo: relayEndpoint,
// },
// }, &node)
// if err != nil {
// logger.Log(1, "failed to send proxy update for relayed node: ", err.Error())
// }
err = mq.NodeUpdate(&relayedNode)
if err != nil {
logger.Log(1, "error sending update to relayed node ", relayedNode.Name, "on network", relay.NetID, ": ", err.Error())
}
}
// send proxy update for node that is relaying traffic
logger.Log(0, "--------> sending relay update to proxy")
err = mq.ProxyUpdate(&manager.ManagerAction{
Action: manager.RelayPeers,
Payload: manager.ManagerPayload{
IsRelay: true,
RelayedPeers: relayPeersMap,
},
}, &node)
if err != nil {
logger.Log(1, "failed to send proxy update: ", err.Error())
}
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(node)
runUpdates(&node, true)

143
logic/peers.go
View File

@@ -14,7 +14,6 @@ import (
"github.com/gravitl/netmaker/logger"
"github.com/gravitl/netmaker/logic/acls/nodeacls"
"github.com/gravitl/netmaker/models"
"github.com/gravitl/netmaker/netclient/ncutils"
"github.com/gravitl/netmaker/servercfg"
"golang.org/x/exp/slices"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
@@ -97,9 +96,9 @@ func GetPeerUpdate(node *models.Node) (models.PeerUpdate, error) {
return models.PeerUpdate{}, err
}
if node.IsRelayed == "yes" {
return GetPeerUpdateForRelayedNode(node, udppeers)
}
// if node.IsRelayed == "yes" {
// return GetPeerUpdateForRelayedNode(node, udppeers)
// }
// #1 Set Keepalive values: set_keepalive
// #2 Set local address: set_local - could be a LOT BETTER and fix some bugs with additional logic
@@ -121,15 +120,15 @@ func GetPeerUpdate(node *models.Node) (models.PeerUpdate, error) {
// if the node is not a server, set the endpoint
var setEndpoint = !(node.IsServer == "yes")
if peer.IsRelayed == "yes" {
if !(node.IsRelay == "yes" && ncutils.StringSliceContains(node.RelayAddrs, peer.PrimaryAddress())) {
//skip -- will be added to relay
continue
} else if node.IsRelay == "yes" && ncutils.StringSliceContains(node.RelayAddrs, peer.PrimaryAddress()) {
// dont set peer endpoint if it's relayed by node
setEndpoint = false
}
}
// if peer.IsRelayed == "yes" {
// if !(node.IsRelay == "yes" && ncutils.StringSliceContains(node.RelayAddrs, peer.PrimaryAddress())) {
// //skip -- will be added to relay
// continue
// } else if node.IsRelay == "yes" && ncutils.StringSliceContains(node.RelayAddrs, peer.PrimaryAddress()) {
// // dont set peer endpoint if it's relayed by node
// setEndpoint = false
// }
// }
if !nodeacls.AreNodesAllowed(nodeacls.NetworkID(node.Network), nodeacls.NodeID(node.ID), nodeacls.NodeID(peer.ID)) {
//skip if not permitted by acl
continue
@@ -238,6 +237,16 @@ func GetPeerUpdate(node *models.Node) (models.PeerUpdate, error) {
peerUpdate.ServerAddrs = serverNodeAddresses
peerUpdate.DNS = getPeerDNS(node.Network)
peerUpdate.PeerIDs = peerMap
if node.IsRelayed == "yes" {
relayNode := FindRelay(node)
relayEndpoint, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", relayNode.Endpoint, relayNode.LocalListenPort))
if err != nil {
logger.Log(1, "failed to resolve relay node endpoint: ", err.Error())
}
peerUpdate.IsRelayed = true
peerUpdate.RelayTo = relayEndpoint
}
return peerUpdate, nil
}
@@ -338,60 +347,60 @@ func GetAllowedIPs(node, peer *models.Node, metrics *models.Metrics) []net.IPNet
}
}
// handle relay gateway peers
if peer.IsRelay == "yes" {
for _, ip := range peer.RelayAddrs {
//find node ID of relayed peer
relayedPeer, err := findNode(ip)
if err != nil {
logger.Log(0, "failed to find node for ip ", ip, err.Error())
continue
}
if relayedPeer == nil {
continue
}
if relayedPeer.ID == node.ID {
//skip self
continue
}
//check if acl permits comms
if !nodeacls.AreNodesAllowed(nodeacls.NetworkID(node.Network), nodeacls.NodeID(node.ID), nodeacls.NodeID(relayedPeer.ID)) {
continue
}
if iplib.Version(net.ParseIP(ip)) == 4 {
relayAddr := net.IPNet{
IP: net.ParseIP(ip),
Mask: net.CIDRMask(32, 32),
}
allowedips = append(allowedips, relayAddr)
}
if iplib.Version(net.ParseIP(ip)) == 6 {
relayAddr := net.IPNet{
IP: net.ParseIP(ip),
Mask: net.CIDRMask(128, 128),
}
allowedips = append(allowedips, relayAddr)
}
relayedNode, err := findNode(ip)
if err != nil {
logger.Log(1, "unable to find node for relayed address", ip, err.Error())
continue
}
if relayedNode.IsEgressGateway == "yes" {
extAllowedIPs := getEgressIPs(node, relayedNode)
allowedips = append(allowedips, extAllowedIPs...)
}
if relayedNode.IsIngressGateway == "yes" {
extPeers, _, err := getExtPeers(relayedNode)
if err == nil {
for _, extPeer := range extPeers {
allowedips = append(allowedips, extPeer.AllowedIPs...)
}
} else {
logger.Log(0, "failed to retrieve extclients from relayed ingress", err.Error())
}
}
}
}
// if peer.IsRelay == "yes" {
// for _, ip := range peer.RelayAddrs {
// //find node ID of relayed peer
// relayedPeer, err := findNode(ip)
// if err != nil {
// logger.Log(0, "failed to find node for ip ", ip, err.Error())
// continue
// }
// if relayedPeer == nil {
// continue
// }
// if relayedPeer.ID == node.ID {
// //skip self
// continue
// }
// //check if acl permits comms
// if !nodeacls.AreNodesAllowed(nodeacls.NetworkID(node.Network), nodeacls.NodeID(node.ID), nodeacls.NodeID(relayedPeer.ID)) {
// continue
// }
// if iplib.Version(net.ParseIP(ip)) == 4 {
// relayAddr := net.IPNet{
// IP: net.ParseIP(ip),
// Mask: net.CIDRMask(32, 32),
// }
// allowedips = append(allowedips, relayAddr)
// }
// if iplib.Version(net.ParseIP(ip)) == 6 {
// relayAddr := net.IPNet{
// IP: net.ParseIP(ip),
// Mask: net.CIDRMask(128, 128),
// }
// allowedips = append(allowedips, relayAddr)
// }
// relayedNode, err := findNode(ip)
// if err != nil {
// logger.Log(1, "unable to find node for relayed address", ip, err.Error())
// continue
// }
// if relayedNode.IsEgressGateway == "yes" {
// extAllowedIPs := getEgressIPs(node, relayedNode)
// allowedips = append(allowedips, extAllowedIPs...)
// }
// if relayedNode.IsIngressGateway == "yes" {
// extPeers, _, err := getExtPeers(relayedNode)
// if err == nil {
// for _, extPeer := range extPeers {
// allowedips = append(allowedips, extPeer.AllowedIPs...)
// }
// } else {
// logger.Log(0, "failed to retrieve extclients from relayed ingress", err.Error())
// }
// }
// }
// }
return allowedips
}

17
logic/relay.go
View File

@@ -72,6 +72,23 @@ func SetRelayedNodes(setRelayed bool, networkName string, addrs []string) ([]mod
}
return returnnodes, nil
}
func GetRelayedNodes(relayNode *models.Node) ([]models.Node, error) {
var returnnodes []models.Node
networkNodes, err := GetNetworkNodes(relayNode.Network)
if err != nil {
return returnnodes, err
}
for _, node := range networkNodes {
if node.IsServer != "yes" {
for _, addr := range relayNode.RelayAddrs {
if addr == node.Address || addr == node.Address6 {
returnnodes = append(returnnodes, node)
}
}
}
}
return returnnodes, nil
}
// ValidateRelay - checks if relay is valid
func ValidateRelay(relay models.RelayRequest) error {

36
logic/server.go
View File

@@ -176,12 +176,40 @@ func ServerJoin(networkSettings *models.Network) (models.Node, error) {
return returnNode, err
}
logger.Log(0, "--------> Hereeeeeee23333")
ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: manager.ManagerPayload{
proxyPayload := manager.ManagerPayload{
IsRelay: node.IsRelay == "yes",
InterfaceName: node.Interface,
Peers: peers.Peers,
},
}
// if proxyPayload.IsRelayed {
// relayNode := FindRelay(node)
// relayEndpoint, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", relayNode.Endpoint, relayNode.LocalListenPort))
// if err != nil {
// logger.Log(1, "failed to resolve relay node endpoint: ", err.Error())
// proxyPayload.IsRelayed = false
// }
// proxyPayload.RelayedTo = relayEndpoint
// }
if proxyPayload.IsRelay {
relayedNodes, err := GetRelayedNodes(node)
if err != nil {
logger.Log(1, "failed to relayed nodes: ", node.Name, err.Error())
proxyPayload.IsRelay = false
} else {
relayPeersMap := make(map[string][]wgtypes.PeerConfig)
for _, relayedNode := range relayedNodes {
peers, err := GetPeersForProxy(&relayedNode)
if err == nil {
relayPeersMap[relayedNode.PublicKey] = peers
}
}
proxyPayload.RelayedPeers = relayPeersMap
}
}
ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: proxyPayload,
}
return *node, nil
}

28
logic/wireguard.go
View File

@@ -160,17 +160,35 @@ func setWGConfig(node *models.Node, peerupdate bool) error {
logger.Log(3, "finished setting wg config on server", node.Name)
}
logger.Log(0, "--------> ADD INTERFACE TO PROXY.....")
logger.Log(0, "--------> ADD/Update INTERFACE TO PROXY.....")
peersP, err := GetPeersForProxy(node)
if err != nil {
logger.Log(0, "failed to get peers for proxy: ", err.Error())
} else {
ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: manager.ManagerPayload{
proxyPayload := manager.ManagerPayload{
IsRelay: node.IsRelay == "yes",
InterfaceName: node.Interface,
Peers: peersP,
},
}
if proxyPayload.IsRelay {
relayedNodes, err := GetRelayedNodes(node)
if err != nil {
logger.Log(1, "failed to relayed nodes: ", node.Name, err.Error())
proxyPayload.IsRelay = false
} else {
relayPeersMap := make(map[string][]wgtypes.PeerConfig)
for _, relayedNode := range relayedNodes {
peers, err := GetPeersForProxy(&relayedNode)
if err == nil {
relayPeersMap[relayedNode.PublicKey] = peers
}
}
proxyPayload.RelayedPeers = relayPeersMap
}
}
ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: proxyPayload,
}
}
return nil

8
models/mqtt.go
View File

@@ -1,6 +1,10 @@
package models
import "golang.zx2c4.com/wireguard/wgctrl/wgtypes"
import (
"net"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
)
// PeerUpdate - struct
type PeerUpdate struct {
@@ -10,6 +14,8 @@ type PeerUpdate struct {
Peers []wgtypes.PeerConfig `json:"peers" bson:"peers" yaml:"peers"`
DNS string `json:"dns" bson:"dns" yaml:"dns"`
PeerIDs PeerMap `json:"peerids" bson:"peerids" yaml:"peerids"`
IsRelayed bool `json:"is_relayed" bson:"is_relayed" yaml:"is_relayed"`
RelayTo *net.UDPAddr `json:"relay_to" bson:"relay_to" yaml:"relay_to"`
}
// KeyUpdate - key update struct

23
mq/publishers.go
View File

@@ -10,6 +10,7 @@ import (
"github.com/gravitl/netmaker/logic"
"github.com/gravitl/netmaker/logic/metrics"
"github.com/gravitl/netmaker/models"
"github.com/gravitl/netmaker/nm-proxy/manager"
"github.com/gravitl/netmaker/servercfg"
"github.com/gravitl/netmaker/serverctl"
)
@@ -106,6 +107,28 @@ func NodeUpdate(node *models.Node) error {
return nil
}
//ProxyUpdate -- publishes updates related to proxy
func ProxyUpdate(proxyPayload *manager.ManagerAction, node *models.Node) error {
if !servercfg.IsMessageQueueBackend() {
return nil
}
if node.IsServer == "yes" {
logic.ProxyMgmChan <- proxyPayload
}
logger.Log(3, "publishing proxy update to "+node.Name)
data, err := json.Marshal(proxyPayload)
if err != nil {
logger.Log(2, "error marshalling node update ", err.Error())
return err
}
if err = publish(node, fmt.Sprintf("update/proxy/%s/%s", node.Network, node.ID), data); err != nil {
logger.Log(2, "error publishing node update to peer ", node.ID, err.Error())
return err
}
return nil
}
// sendPeers - retrieve networks, send peer ports to all peers
func sendPeers() {

8
netclient/functions/daemon.go
View File

@@ -217,6 +217,14 @@ func setSubscriptions(client mqtt.Client, nodeCfg *config.ClientConfig) {
}
return
}
if token := client.Subscribe(fmt.Sprintf("update/proxy/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(ProxyUpdate)); token.WaitTimeout(mq.MQ_TIMEOUT*time.Second) && token.Error() != nil {
if token.Error() == nil {
logger.Log(0, "network:", nodeCfg.Node.Network, "connection timeout")
} else {
logger.Log(0, "network:", nodeCfg.Node.Network, token.Error().Error())
}
return
}
logger.Log(3, fmt.Sprintf("subscribed to node updates for node %s update/%s/%s", nodeCfg.Node.Name, nodeCfg.Node.Network, nodeCfg.Node.ID))
if token := client.Subscribe(fmt.Sprintf("peers/%s/%s", nodeCfg.Node.Network, nodeCfg.Node.ID), 0, mqtt.MessageHandler(UpdatePeers)); token.Wait() && token.Error() != nil {
logger.Log(0, "network", nodeCfg.Node.Network, token.Error().Error())

28
netclient/functions/mqhandlers.go
View File

@@ -31,6 +31,26 @@ var All mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
//logger.Log(0, "Message: " + string(msg.Payload()))
}
func ProxyUpdate(client mqtt.Client, msg mqtt.Message) {
var nodeCfg config.ClientConfig
var proxyUpdate manager.ManagerAction
var network = strings.Split(msg.Topic(), "/")[2]
nodeCfg.Network = network
nodeCfg.ReadConfig()
data, dataErr := decryptMsg(&nodeCfg, msg.Payload())
if dataErr != nil {
return
}
err := json.Unmarshal([]byte(data), &proxyUpdate)
if err != nil {
logger.Log(0, "error unmarshalling node update data"+err.Error())
return
}
logger.Log(0, "---------> recieved a proxy update")
ProxyMgmChan <- &proxyUpdate
}
// NodeUpdate -- mqtt message handler for /update/<NodeID> topic
func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
var newNode models.Node
@@ -145,6 +165,12 @@ func NodeUpdate(client mqtt.Client, msg mqtt.Message) {
// }
// }
// }
ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: manager.ManagerPayload{
IsRelayed: newNode.IsRelay == "yes",
},
}
if ifaceDelta { // if a change caused an ifacedelta we need to notify the server to update the peers
doneErr := publishSignal(&nodeCfg, ncutils.DONE)
if doneErr != nil {
@@ -252,6 +278,8 @@ func UpdatePeers(client mqtt.Client, msg mqtt.Message) {
Payload: manager.ManagerPayload{
InterfaceName: cfg.Node.Interface,
Peers: peerUpdate.Peers,
IsRelayed: peerUpdate.IsRelayed,
RelayedTo: peerUpdate.RelayTo,
},
}
logger.Log(0, "network:", cfg.Node.Network, "received peer update for node "+cfg.Node.Name+" "+cfg.Node.Network)

7
nm-proxy/common/common.go
View File

@@ -23,6 +23,7 @@ import (
)
var IsHostNetwork bool
var IsRelay bool
const (
NmProxyPort = 51722
@@ -51,6 +52,7 @@ type Config struct {
BodySize int
Addr string
RemoteKey string
LocalKey string
WgInterface *wg.WGIface
AllowedIps []net.IPNet
PreSharedKey *wgtypes.Key
@@ -69,12 +71,17 @@ type Proxy struct {
type RemotePeer struct {
PeerKey string
Interface string
Endpoint *net.UDPAddr
}
var WgIFaceMap = make(map[string]map[string]*Conn)
var PeerKeyHashMap = make(map[string]RemotePeer)
var WgIfaceKeyMap = make(map[string]struct{})
var RelayPeerMap = make(map[string]map[string]RemotePeer)
// RunCmd - runs a local command
func RunCmd(command string, printerr bool) (string, error) {
args := strings.Fields(command)

57
nm-proxy/manager/manager.go
View File

@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"log"
"net"
"runtime"
"github.com/gravitl/netmaker/netclient/wireguard"
@@ -19,12 +20,19 @@ type ProxyAction string
type ManagerPayload struct {
InterfaceName string
Peers []wgtypes.PeerConfig
IsRelayed bool
RelayedTo *net.UDPAddr
IsRelay bool
RelayedPeers map[string][]wgtypes.PeerConfig
}
const (
AddInterface ProxyAction = "ADD_INTERFACE"
DeletePeer ProxyAction = "DELETE_PEER"
UpdatePeer ProxyAction = "UPDATE_PEER"
RelayPeers ProxyAction = "RELAY_PEERS"
RelayUpdate ProxyAction = "RELAY_UPDATE"
RelayTo ProxyAction = "RELAY_TO"
)
type ManagerAction struct {
@@ -45,23 +53,42 @@ func StartProxyManager(manageChan chan *ManagerAction) {
log.Printf("failed to add interface: [%s] to proxy: %v\n ", mI.Payload.InterfaceName, err)
}
case UpdatePeer:
mI.UpdatePeerProxy()
//mI.UpdatePeerProxy()
case DeletePeer:
mI.DeletePeers()
case RelayPeers:
mI.RelayPeers()
case RelayUpdate:
mI.RelayUpdate()
}
}
}
}
func cleanUp(iface string) {
if peers, ok := common.WgIFaceMap[iface]; ok {
log.Println("########------------> CLEANING UP: ", iface)
for _, peerI := range peers {
peerI.Proxy.Cancel()
func (m *ManagerAction) RelayUpdate() {
common.IsRelay = m.Payload.IsRelay
}
func (m *ManagerAction) RelayPeers() {
common.IsRelay = true
for relayedNodePubKey, peers := range m.Payload.RelayedPeers {
for _, peer := range peers {
if _, ok := common.RelayPeerMap[relayedNodePubKey]; !ok {
common.RelayPeerMap[relayedNodePubKey] = make(map[string]common.RemotePeer)
}
peer.Endpoint.Port = common.NmProxyPort
relayedNodePubKeyHash := fmt.Sprintf("%x", md5.Sum([]byte(relayedNodePubKey)))
remotePeerKeyHash := fmt.Sprintf("%x", md5.Sum([]byte(peer.PublicKey.String())))
if _, ok := common.RelayPeerMap[relayedNodePubKeyHash]; !ok {
common.RelayPeerMap[relayedNodePubKeyHash] = make(map[string]common.RemotePeer)
}
common.RelayPeerMap[relayedNodePubKeyHash][remotePeerKeyHash] = common.RemotePeer{
Endpoint: peer.Endpoint,
}
}
delete(common.WgIFaceMap, iface)
}
}
func (m *ManagerAction) DeletePeers() {
@@ -112,6 +139,16 @@ func (m *ManagerAction) UpdatePeerProxy() {
}
func cleanUp(iface string) {
if peers, ok := common.WgIFaceMap[iface]; ok {
log.Println("########------------> CLEANING UP: ", iface)
for _, peerI := range peers {
peerI.Proxy.Cancel()
}
}
delete(common.WgIFaceMap, iface)
}
func (m *ManagerAction) AddInterfaceToProxy() error {
var err error
if m.Payload.InterfaceName == "" {
@@ -133,7 +170,8 @@ func (m *ManagerAction) AddInterfaceToProxy() error {
wgInterface, err := wg.NewWGIFace(ifaceName, "127.0.0.1/32", wg.DefaultMTU)
if err != nil {
log.Fatal("Failed init new interface: ", err)
log.Println("Failed init new interface: ", err)
return err
}
log.Printf("wg: %+v\n", wgInterface)
@@ -146,7 +184,8 @@ func (m *ManagerAction) AddInterfaceToProxy() error {
Interface: ifaceName,
PeerKey: peerI.PublicKey.String(),
}
peerpkg.AddNewPeer(wgInterface, &peerI)
peerpkg.AddNewPeer(wgInterface, &peerI, m.Payload.IsRelayed, m.Payload.RelayedTo)
}
log.Printf("------> PEERHASHMAP: %+v\n", common.PeerKeyHashMap)
return nil

51
nm-proxy/packet/packet.go
View File

@@ -1,45 +1,36 @@
package packet
import (
"bytes"
"crypto/md5"
"encoding/binary"
"fmt"
"log"
)
var udpHeaderLen = 8
func ProcessPacketBeforeSending(buf []byte, srckey string, n, dstPort int) ([]byte, int, string, error) {
portbuf := new(bytes.Buffer)
binary.Write(portbuf, binary.BigEndian, uint16(dstPort))
hmd5 := md5.Sum([]byte(srckey))
if n > len(buf)-18 {
buf = append(buf, portbuf.Bytes()[0])
buf = append(buf, portbuf.Bytes()[1])
buf = append(buf, hmd5[:]...)
func ProcessPacketBeforeSending(buf []byte, n int, srckey, dstKey string) ([]byte, int, string, string) {
srcKeymd5 := md5.Sum([]byte(srckey))
dstKeymd5 := md5.Sum([]byte(dstKey))
if n > len(buf)-len(srcKeymd5)-len(dstKeymd5) {
buf = append(buf, srcKeymd5[:]...)
buf = append(buf, dstKeymd5[:]...)
} else {
buf[n] = portbuf.Bytes()[0]
buf[n+1] = portbuf.Bytes()[1]
copy(buf[n+2:n+2+len(hmd5)], hmd5[:])
copy(buf[n:n+len(srcKeymd5)], srcKeymd5[:])
copy(buf[n+len(srcKeymd5):n+len(srcKeymd5)+len(dstKeymd5)], dstKeymd5[:])
}
n += len(srcKeymd5)
n += len(dstKeymd5)
return buf, n, fmt.Sprintf("%x", srcKeymd5), fmt.Sprintf("%x", dstKeymd5)
}
n += 2
n += len(hmd5)
return buf, n, fmt.Sprintf("%x", hmd5), nil
}
func ExtractInfo(buffer []byte, n int) (int, int, string, error) {
func ExtractInfo(buffer []byte, n int) (int, string, string) {
data := buffer[:n]
var localWgPort uint16
portBuf := data[n-18 : n-18+3]
keyHash := data[n-16:]
reader := bytes.NewReader(portBuf)
err := binary.Read(reader, binary.BigEndian, &localWgPort)
if err != nil {
log.Println("Failed to read port buffer: ", err)
if len(data) < 32 {
return 0, "", ""
}
n -= 18
return int(localWgPort), n, fmt.Sprintf("%x", keyHash), err
srcKeyHash := data[n-32 : n-16]
dstKeyHash := data[n-16:]
n -= 32
return n, fmt.Sprintf("%x", srcKeyHash), fmt.Sprintf("%x", dstKeyHash)
}

19
nm-proxy/peer/peer.go
View File

@@ -1,12 +1,14 @@
package peer
import (
"crypto/md5"
"fmt"
"log"
"net"
"github.com/gravitl/netmaker/nm-proxy/common"
"github.com/gravitl/netmaker/nm-proxy/proxy"
"github.com/gravitl/netmaker/nm-proxy/server"
"github.com/gravitl/netmaker/nm-proxy/wg"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
)
@@ -32,16 +34,23 @@ type ConnConfig struct {
RemoteProxyPort int
}
func AddNewPeer(wgInterface *wg.WGIface, peer *wgtypes.PeerConfig) error {
func AddNewPeer(wgInterface *wg.WGIface, peer *wgtypes.PeerConfig, isRelayed bool, relayTo *net.UDPAddr) error {
c := proxy.Config{
Port: peer.Endpoint.Port,
LocalKey: wgInterface.Device.PublicKey.String(),
RemoteKey: peer.PublicKey.String(),
WgInterface: wgInterface,
AllowedIps: peer.AllowedIPs,
}
p := proxy.NewProxy(c)
remoteConn, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", peer.Endpoint.IP.String(), common.NmProxyPort))
peerEndpoint := peer.Endpoint.IP.String()
if isRelayed {
go server.NmProxyServer.KeepAlive(peer.Endpoint.IP.String(), common.NmProxyPort)
peerEndpoint = relayTo.IP.String()
}
remoteConn, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", peerEndpoint, common.NmProxyPort))
if err != nil {
return err
}
@@ -59,11 +68,13 @@ func AddNewPeer(wgInterface *wg.WGIface, peer *wgtypes.PeerConfig) error {
RemoteWgPort: peer.Endpoint.Port,
RemoteProxyPort: common.NmProxyPort,
}
peerProxy := common.Proxy{
Ctx: p.Ctx,
Cancel: p.Cancel,
Config: common.Config{
Port: peer.Endpoint.Port,
LocalKey: wgInterface.Device.PublicKey.String(),
RemoteKey: peer.PublicKey.String(),
WgInterface: wgInterface,
AllowedIps: peer.AllowedIPs,
@@ -72,6 +83,9 @@ func AddNewPeer(wgInterface *wg.WGIface, peer *wgtypes.PeerConfig) error {
RemoteConn: remoteConn,
LocalConn: p.LocalConn,
}
if isRelayed {
connConf.RemoteProxyIP = relayTo.IP
}
peerConn := common.Conn{
Config: connConf,
Proxy: peerProxy,
@@ -82,5 +96,6 @@ func AddNewPeer(wgInterface *wg.WGIface, peer *wgtypes.PeerConfig) error {
common.WgIFaceMap[wgInterface.Name] = make(map[string]*common.Conn)
common.WgIFaceMap[wgInterface.Name][peer.PublicKey.String()] = &peerConn
}
common.WgIfaceKeyMap[fmt.Sprintf("%x", md5.Sum([]byte(wgInterface.Device.PublicKey.String())))] = struct{}{}
return nil
}

14
nm-proxy/proxy/wireguard.go
View File

@@ -58,13 +58,13 @@ func (p *Proxy) ProxyToRemote() {
}
peers := common.WgIFaceMap[p.Config.WgInterface.Name]
if peerI, ok := peers[p.Config.RemoteKey]; ok {
var srcPeerHash string
buf, n, srcPeerHash, err = packet.ProcessPacketBeforeSending(buf, peerI.Config.LocalKey, n, peerI.Config.RemoteWgPort)
var srcPeerKeyHash, dstPeerKeyHash string
buf, n, srcPeerKeyHash, dstPeerKeyHash = packet.ProcessPacketBeforeSending(buf, n, peerI.Config.LocalKey, peerI.Config.Key)
if err != nil {
log.Println("failed to process pkt before sending: ", err)
}
log.Printf("PROXING TO REMOTE!!!---> %s >>>>> %s [[ DstPort: %d, SrcPeerHash: %x ]]\n",
server.NmProxyServer.Server.LocalAddr().String(), p.RemoteConn.String(), peerI.Config.RemoteWgPort, srcPeerHash)
log.Printf("PROXING TO REMOTE!!!---> %s >>>>> %s [[ DstPort: %d, SrcPeerHash: %s, DstPeerHash: %s ]]\n",
server.NmProxyServer.Server.LocalAddr().String(), p.RemoteConn.String(), peerI.Config.RemoteWgPort, srcPeerKeyHash, dstPeerKeyHash)
} else {
log.Printf("Peer: %s not found in config\n", p.Config.RemoteKey)
continue
@@ -78,6 +78,12 @@ func (p *Proxy) ProxyToRemote() {
}
}
}
func test(n int, buffer []byte) {
data := buffer[:n]
srcKeyHash := data[n-32 : n-16]
dstKeyHash := data[n-16:]
log.Printf("--------> TEST PACKET [ SRCKEYHASH: %x ], [ DSTKEYHASH: %x ] \n", srcKeyHash, dstKeyHash)
}
func (p *Proxy) updateEndpoint() error {
udpAddr, err := net.ResolveUDPAddr("udp", p.LocalConn.LocalAddr().String())

46
nm-proxy/server/server.go
View File

@@ -22,6 +22,7 @@ const (
type Config struct {
Port int
BodySize int
IsRelay bool
Addr net.Addr
}
@@ -34,7 +35,7 @@ type ProxyServer struct {
func (p *ProxyServer) Listen() {
// Buffer with indicated body size
buffer := make([]byte, 1502)
buffer := make([]byte, 1532)
for {
// Read Packet
n, source, err := p.Server.ReadFromUDP(buffer)
@@ -42,28 +43,40 @@ func (p *ProxyServer) Listen() {
log.Println("RECV ERROR: ", err)
continue
}
var localWgPort int
var srcPeerKeyHash string
localWgPort, n, srcPeerKeyHash, err = packet.ExtractInfo(buffer, n)
if err != nil {
log.Println("failed to extract info: ", err)
continue
}
var srcPeerKeyHash, dstPeerKeyHash string
n, srcPeerKeyHash, dstPeerKeyHash = packet.ExtractInfo(buffer, n)
//log.Printf("--------> RECV PKT [DSTPORT: %d], [SRCKEYHASH: %s], SourceIP: [%s] \n", localWgPort, srcPeerKeyHash, source.IP.String())
if common.IsRelay && dstPeerKeyHash != "" {
if _, ok := common.WgIfaceKeyMap[dstPeerKeyHash]; !ok {
log.Println("----------> Relaying######")
// check for routing map and forward to right proxy
if remoteMap, ok := common.RelayPeerMap[srcPeerKeyHash]; ok {
if conf, ok := remoteMap[dstPeerKeyHash]; ok {
log.Printf("--------> Relaying PKT [ SourceIP: %s:%d ], [ SourceKeyHash: %s ], [ DstIP: %s:%d ], [ DstHashKey: %s ] \n",
source.IP.String(), source.Port, srcPeerKeyHash, conf.Endpoint.String(), conf.Endpoint.Port, dstPeerKeyHash)
_, err = NmProxyServer.Server.WriteToUDP(buffer[:n+32], conf.Endpoint)
if err != nil {
log.Println("Failed to send to remote: ", err)
}
}
}
}
}
if peerInfo, ok := common.PeerKeyHashMap[srcPeerKeyHash]; ok {
if peers, ok := common.WgIFaceMap[peerInfo.Interface]; ok {
if peerI, ok := peers[peerInfo.PeerKey]; ok {
// if peerI.Config.LocalWgPort == int(localWgPort) {
log.Printf("PROXING TO LOCAL!!!---> %s <<<< %s <<<<<<<< %s [[ RECV PKT [DSTPORT: %d], [SRCKEYHASH: %s], SourceIP: [%s] ]]\n",
log.Printf("PROXING TO LOCAL!!!---> %s <<<< %s <<<<<<<< %s [[ RECV PKT [SRCKEYHASH: %s], [DSTKEYHASH: %s], SourceIP: [%s] ]]\n",
peerI.Proxy.LocalConn.RemoteAddr(), peerI.Proxy.LocalConn.LocalAddr(),
fmt.Sprintf("%s:%d", source.IP.String(), source.Port), localWgPort, srcPeerKeyHash, source.IP.String())
fmt.Sprintf("%s:%d", source.IP.String(), source.Port), srcPeerKeyHash, dstPeerKeyHash, source.IP.String())
_, err = peerI.Proxy.LocalConn.Write(buffer[:n])
if err != nil {
log.Println("Failed to proxy to Wg local interface: ", err)
continue
}
// }
}
}
@@ -93,9 +106,12 @@ func (p *ProxyServer) CreateProxyServer(port, bodySize int, addr string) (err er
func (p *ProxyServer) KeepAlive(ip string, port int) {
for {
_, _ = p.Server.Write([]byte("hello-proxy"))
//fmt.Println("Sending MSg: ", err)
time.Sleep(time.Second)
_, _ = p.Server.WriteToUDP([]byte("hello-proxy"), &net.UDPAddr{
IP: net.ParseIP(ip),
Port: port,
})
//log.Println("Sending MSg: ", ip, port, err)
time.Sleep(time.Second * 5)
}
}

19
nm-proxy/wg/wg.go
View File

@@ -53,7 +53,10 @@ func NewWGIFace(iface string, address string, mtu int) (*WGIface, error) {
}
wgIface.Address = wgAddress
wgIface.GetWgIface(iface)
err = wgIface.GetWgIface(iface)
if err != nil {
return nil, err
}
return wgIface, nil
}
@@ -73,6 +76,20 @@ func (w *WGIface) GetWgIface(iface string) error {
return nil
}
func GetWgIfacePubKey(iface string) string {
wgClient, err := wgctrl.New()
if err != nil {
log.Println("Error fetching pub key: ", iface, err)
return ""
}
dev, err := wgClient.Device(iface)
if err != nil {
log.Println("Error fetching pub key: ", iface, err)
return ""
}
return dev.PublicKey.String()
}
// parseAddress parse a string ("1.2.3.4/24") address to WG Address
func parseAddress(address string) (WGAddress, error) {
ip, network, err := net.ParseCIDR(address)

27
serverctl/serverctl.go
View File

@@ -14,6 +14,7 @@ import (
"github.com/gravitl/netmaker/netclient/ncutils"
"github.com/gravitl/netmaker/nm-proxy/manager"
"github.com/gravitl/netmaker/servercfg"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
)
const (
@@ -88,12 +89,30 @@ func SyncServerNetworkWithProxy() error {
continue
}
logger.Log(0, "----> HEREEEEEEEE1")
proxyPayload := manager.ManagerPayload{
IsRelay: serverNode.IsRelay == "yes",
InterfaceName: serverNode.Interface,
Peers: peers,
}
if proxyPayload.IsRelay {
relayedNodes, err := logic.GetRelayedNodes(&serverNode)
if err != nil {
logger.Log(1, "failed to relayed nodes: ", serverNode.Name, err.Error())
proxyPayload.IsRelay = false
} else {
relayPeersMap := make(map[string][]wgtypes.PeerConfig)
for _, relayedNode := range relayedNodes {
peers, err := logic.GetPeersForProxy(&relayedNode)
if err == nil {
relayPeersMap[relayedNode.PublicKey] = peers
}
}
proxyPayload.RelayedPeers = relayPeersMap
}
}
logic.ProxyMgmChan <- &manager.ManagerAction{
Action: manager.AddInterface,
Payload: manager.ManagerPayload{
InterfaceName: serverNetworkSettings.DefaultInterface,
Peers: peers,
},
Payload: proxyPayload,
}
}