sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-09-22 06:39:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

Initial code

Browse Source
This commit is contained in:
Zoltán Papp
2024-04-12 17:38:31 +02:00
parent 76702c8a09
commit d67f766b2e
7 changed files with 418 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/internal/engine.go
+50 -7
View File
@@ -138,6 +138,7 @@ type Engine struct {
signalProbe *Probe
relayProbe *Probe
wgProbe *Probe
turnRelay *relay.PermanentTurn
}
// Peer is an instance of the Connection Peer
@@ -452,10 +453,19 @@ func SignalOfferAnswer(offerAnswer peer.OfferAnswer, myKey wgtypes.Key, remoteKe
t = sProto.Body_OFFER
}
msg, err := signal.MarshalCredential(myKey, offerAnswer.WgListenPort, remoteKey, &signal.Credential{
UFrag: offerAnswer.IceCredentials.UFrag,
Pwd: offerAnswer.IceCredentials.Pwd,
}, t, offerAnswer.RosenpassPubKey, offerAnswer.RosenpassAddr)
msg, err := signal.MarshalCredential(
myKey,
offerAnswer.WgListenPort,
remoteKey, &signal.Credential{
UFrag: offerAnswer.IceCredentials.UFrag,
Pwd: offerAnswer.IceCredentials.Pwd,
},
t,
offerAnswer.RosenpassPubKey,
offerAnswer.RosenpassAddr,
offerAnswer.RelayedAddr.String(),
offerAnswer.RemoteAddr.String(),
)
if err != nil {
return err
}
@@ -483,6 +493,13 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
return err
}
turnRelay := relay.NewPermanentTurn(e.STUNs[0], e.TURNs[0])
err = turnRelay.Open()
if err != nil {
return fmt.Errorf("faile to open turn relay: %w", err)
}
e.turnRelay = turnRelay
// todo update signal
}
@@ -603,6 +620,7 @@ func (e *Engine) updateSTUNs(stuns []*mgmProto.HostConfig) error {
var newSTUNs []*stun.URI
log.Debugf("got STUNs update from Management Service, updating")
for _, s := range stuns {
log.Debugf("-----updated TURN: %s", s.Uri)
url, err := stun.ParseURI(s.Uri)
if err != nil {
return err
@@ -621,6 +639,7 @@ func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error {
var newTURNs []*stun.URI
log.Debugf("got TURNs update from Management Service, updating")
for _, turn := range turns {
log.Debugf("-----updated Turn %v, %s, %s", turn.HostConfig.Uri, turn.User, turn.Password)
url, err := stun.ParseURI(turn.HostConfig.Uri)
if err != nil {
return err
@@ -630,7 +649,6 @@ func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error {
newTURNs = append(newTURNs, url)
}
e.TURNs = newTURNs
return nil
}
@@ -934,7 +952,7 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs string) (*peer.Conn, e
RosenpassAddr: e.getRosenpassAddr(),
}
peerConn, err := peer.NewConn(config, e.statusRecorder, e.wgProxyFactory, e.mobileDep.TunAdapter, e.mobileDep.IFaceDiscover)
peerConn, err := peer.NewConn(config, e.statusRecorder, e.wgProxyFactory, e.mobileDep.TunAdapter, e.mobileDep.IFaceDiscover, e.turnRelay)
if err != nil {
return nil, err
}
@@ -1000,6 +1018,17 @@ func (e *Engine) receiveSignalEvents() {
rosenpassPubKey = msg.GetBody().GetRosenpassConfig().GetRosenpassPubKey()
rosenpassAddr = msg.GetBody().GetRosenpassConfig().GetRosenpassServerAddr()
}
relayedAddr, err := net.ResolveUDPAddr("udp", msg.GetBody().GetRelay().GetRelayedAddress())
if err != nil {
return err
}
remoteAddr, err := net.ResolveUDPAddr("udp", msg.GetBody().GetRelay().GetSrvRefAddress())
if err != nil {
return err
}
conn.OnRemoteOffer(peer.OfferAnswer{
IceCredentials: peer.IceCredentials{
UFrag: remoteCred.UFrag,
@@ -1009,6 +1038,8 @@ func (e *Engine) receiveSignalEvents() {
Version: msg.GetBody().GetNetBirdVersion(),
RosenpassPubKey: rosenpassPubKey,
RosenpassAddr: rosenpassAddr,
RelayedAddr: relayedAddr,
RemoteAddr: remoteAddr,
})
case sProto.Body_ANSWER:
remoteCred, err := signal.UnMarshalCredential(msg)
@@ -1024,6 +1055,17 @@ func (e *Engine) receiveSignalEvents() {
rosenpassPubKey = msg.GetBody().GetRosenpassConfig().GetRosenpassPubKey()
rosenpassAddr = msg.GetBody().GetRosenpassConfig().GetRosenpassServerAddr()
}
relayedAddr, err := net.ResolveUDPAddr("udp", msg.GetBody().GetRelay().GetRelayedAddress())
if err != nil {
return err
}
remoteAddr, err := net.ResolveUDPAddr("udp", msg.GetBody().GetRelay().GetSrvRefAddress())
if err != nil {
return err
}
conn.OnRemoteAnswer(peer.OfferAnswer{
IceCredentials: peer.IceCredentials{
UFrag: remoteCred.UFrag,
@@ -1033,6 +1075,8 @@ func (e *Engine) receiveSignalEvents() {
Version: msg.GetBody().GetNetBirdVersion(),
RosenpassPubKey: rosenpassPubKey,
RosenpassAddr: rosenpassAddr,
RelayedAddr: relayedAddr,
RemoteAddr: remoteAddr,
})
case sProto.Body_CANDIDATE:
candidate, err := ice.UnmarshalCandidate(msg.GetBody().Payload)
@@ -1043,7 +1087,6 @@ func (e *Engine) receiveSignalEvents() {
conn.OnRemoteCandidate(candidate)
case sProto.Body_MODE:
}
return nil
})
if err != nil {
client/internal/peer/conn.go
+55 -69
View File
@@ -4,7 +4,6 @@ import (
"context"
"fmt"
"net"
"runtime"
"strings"
"sync"
"time"
@@ -14,6 +13,7 @@ import (
log "github.com/sirupsen/logrus"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/internal/relay"
"github.com/netbirdio/netbird/client/internal/stdnet"
"github.com/netbirdio/netbird/client/internal/wgproxy"
"github.com/netbirdio/netbird/iface"
@@ -93,6 +93,10 @@ type OfferAnswer struct {
// RosenpassAddr is the Rosenpass server address (IP:port) of the remote peer when receiving this message
// This value is the local Rosenpass server address when sending the message
RosenpassAddr string
// Turn Relay
RelayedAddr net.Addr
RemoteAddr net.Addr
}
// IceCredentials ICE protocol credentials struct
@@ -141,11 +145,11 @@ type Conn struct {
sentExtraSrflx bool
remoteEndpoint *net.UDPAddr
remoteConn *ice.Conn
connID nbnet.ConnectionID
beforeAddPeerHooks []BeforeAddPeerHookFunc
afterRemovePeerHooks []AfterRemovePeerHookFunc
turnRelay *relay.PermanentTurn
}
// meta holds meta information about a connection
@@ -176,7 +180,7 @@ func (conn *Conn) UpdateStunTurn(turnStun []*stun.URI) {
// NewConn creates a new not opened Conn to the remote peer.
// To establish a connection run Conn.Open
func NewConn(config ConnConfig, statusRecorder *Status, wgProxyFactory *wgproxy.Factory, adapter iface.TunAdapter, iFaceDiscover stdnet.ExternalIFaceDiscover) (*Conn, error) {
func NewConn(config ConnConfig, statusRecorder *Status, wgProxyFactory *wgproxy.Factory, adapter iface.TunAdapter, iFaceDiscover stdnet.ExternalIFaceDiscover, turnRelay *relay.PermanentTurn) (*Conn, error) {
return &Conn{
config: config,
mu: sync.Mutex{},
@@ -189,6 +193,7 @@ func NewConn(config ConnConfig, statusRecorder *Status, wgProxyFactory *wgproxy.
wgProxyFactory: wgProxyFactory,
adapter: adapter,
iFaceDiscover: iFaceDiscover,
turnRelay: turnRelay,
}, nil
}
@@ -212,7 +217,7 @@ func (conn *Conn) reCreateAgent() error {
MulticastDNSMode: ice.MulticastDNSModeDisabled,
NetworkTypes: []ice.NetworkType{ice.NetworkTypeUDP4, ice.NetworkTypeUDP6},
Urls: conn.config.StunTurn,
CandidateTypes: conn.candidateTypes(),
CandidateTypes: []ice.CandidateType{},
FailedTimeout: &failedTimeout,
InterfaceFilter: stdnet.InterfaceFilter(conn.config.InterfaceBlackList),
UDPMux: conn.config.UDPMux,
@@ -262,17 +267,6 @@ func (conn *Conn) reCreateAgent() error {
return nil
}
func (conn *Conn) candidateTypes() []ice.CandidateType {
if hasICEForceRelayConn() {
return []ice.CandidateType{ice.CandidateTypeRelay}
}
// TODO: remove this once we have refactored userspace proxy into the bind package
if runtime.GOOS == "ios" {
return []ice.CandidateType{ice.CandidateTypeHost, ice.CandidateTypeServerReflexive}
}
return []ice.CandidateType{ice.CandidateTypeHost, ice.CandidateTypeServerReflexive, ice.CandidateTypeRelay}
}
// Open opens connection to the remote peer starting ICE candidate gathering process.
// Blocks until connection has been closed or connection timeout.
// ConnStatus will be set accordingly
@@ -351,23 +345,12 @@ func (conn *Conn) Open() error {
log.Warnf("error while updating the state of peer %s,err: %v", conn.config.Key, err)
}
err = conn.agent.GatherCandidates()
if err != nil {
return err
}
// will block until connection succeeded
// but it won't release if ICE Agent went into Disconnected or Failed state,
// so we have to cancel it with the provided context once agent detected a broken connection
isControlling := conn.config.LocalKey > conn.config.Key
var remoteConn *ice.Conn
if isControlling {
remoteConn, err = conn.agent.Dial(conn.ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
} else {
remoteConn, err = conn.agent.Accept(conn.ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
}
if err != nil {
return err
err = conn.turnRelay.PunchHole(remoteOfferAnswer.RemoteAddr)
if err != nil {
log.Errorf("failed to punch hole: %v", err)
}
}
// dynamically set remote WireGuard port is other side specified a different one from the default one
@@ -376,7 +359,11 @@ func (conn *Conn) Open() error {
remoteWgPort = remoteOfferAnswer.WgListenPort
}
conn.remoteConn = remoteConn
// todo configure the wg with proper address
remoteConn, err := net.Dial("udp", remoteOfferAnswer.RemoteAddr.String())
if err != nil {
log.Errorf("failed to dial remote peer %s: %v", conn.config.Key, err)
}
// the ice connection has been established successfully so we are ready to start the proxy
remoteAddr, err := conn.configureConnection(remoteConn, remoteWgPort, remoteOfferAnswer.RosenpassPubKey,
@@ -415,25 +402,14 @@ func (conn *Conn) configureConnection(remoteConn net.Conn, remoteWgPort int, rem
conn.mu.Lock()
defer conn.mu.Unlock()
pair, err := conn.agent.GetSelectedCandidatePair()
var endpoint net.Addr
log.Debugf("setup relay connection")
conn.wgProxy = conn.wgProxyFactory.GetProxy()
endpoint, err := conn.wgProxy.AddTurnConn(remoteConn)
if err != nil {
return nil, err
}
var endpoint net.Addr
if isRelayCandidate(pair.Local) {
log.Debugf("setup relay connection")
conn.wgProxy = conn.wgProxyFactory.GetProxy()
endpoint, err = conn.wgProxy.AddTurnConn(remoteConn)
if err != nil {
return nil, err
}
} else {
// To support old version's with direct mode we attempt to punch an additional role with the remote WireGuard port
go conn.punchRemoteWGPort(pair, remoteWgPort)
endpoint = remoteConn.RemoteAddr()
}
endpointUdpAddr, _ := net.ResolveUDPAddr(endpoint.Network(), endpoint.String())
conn.remoteEndpoint = endpointUdpAddr
log.Debugf("Conn resolved IP for %s: %s", endpoint, endpointUdpAddr.IP)
@@ -454,31 +430,33 @@ func (conn *Conn) configureConnection(remoteConn net.Conn, remoteWgPort int, rem
}
conn.status = StatusConnected
rosenpassEnabled := false
if remoteRosenpassPubKey != nil {
rosenpassEnabled = true
}
/*
rosenpassEnabled := false
if remoteRosenpassPubKey != nil {
rosenpassEnabled = true
}
peerState := State{
PubKey: conn.config.Key,
ConnStatus: conn.status,
ConnStatusUpdate: time.Now(),
LocalIceCandidateType: pair.Local.Type().String(),
RemoteIceCandidateType: pair.Remote.Type().String(),
LocalIceCandidateEndpoint: fmt.Sprintf("%s:%d", pair.Local.Address(), pair.Local.Port()),
RemoteIceCandidateEndpoint: fmt.Sprintf("%s:%d", pair.Remote.Address(), pair.Local.Port()),
Direct: !isRelayCandidate(pair.Local),
RosenpassEnabled: rosenpassEnabled,
Mux: new(sync.RWMutex),
}
if pair.Local.Type() == ice.CandidateTypeRelay || pair.Remote.Type() == ice.CandidateTypeRelay {
peerState.Relayed = true
}
peerState := State{
PubKey: conn.config.Key,
ConnStatus: conn.status,
ConnStatusUpdate: time.Now(),
LocalIceCandidateType: pair.Local.Type().String(),
RemoteIceCandidateType: pair.Remote.Type().String(),
LocalIceCandidateEndpoint: fmt.Sprintf("%s:%d", pair.Local.Address(), pair.Local.Port()),
RemoteIceCandidateEndpoint: fmt.Sprintf("%s:%d", pair.Remote.Address(), pair.Local.Port()),
Direct: !isRelayCandidate(pair.Local),
RosenpassEnabled: rosenpassEnabled,
Mux: new(sync.RWMutex),
}
if pair.Local.Type() == ice.CandidateTypeRelay || pair.Remote.Type() == ice.CandidateTypeRelay {
peerState.Relayed = true
}
err = conn.statusRecorder.UpdatePeerState(peerState)
if err != nil {
log.Warnf("unable to save peer's state, got error: %v", err)
}
err = conn.statusRecorder.UpdatePeerState(peerState)
if err != nil {
log.Warnf("unable to save peer's state, got error: %v", err)
}
*/
_, ipNet, err := net.ParseCIDR(conn.config.WgConfig.AllowedIps)
if err != nil {
@@ -680,6 +658,8 @@ func (conn *Conn) sendAnswer() error {
Version: version.NetbirdVersion(),
RosenpassPubKey: conn.config.RosenpassPubKey,
RosenpassAddr: conn.config.RosenpassAddr,
RelayedAddr: conn.turnRelay.RelayedAddress(),
RemoteAddr: conn.turnRelay.SrvRefAddr(),
})
if err != nil {
return err
@@ -703,6 +683,8 @@ func (conn *Conn) sendOffer() error {
Version: version.NetbirdVersion(),
RosenpassPubKey: conn.config.RosenpassPubKey,
RosenpassAddr: conn.config.RosenpassAddr,
RelayedAddr: conn.turnRelay.RelayedAddress(),
RemoteAddr: conn.turnRelay.SrvRefAddr(),
})
if err != nil {
return err
@@ -742,6 +724,10 @@ func (conn *Conn) Status() ConnStatus {
return conn.status
}
func (conn *Conn) OnRemoteRelayRequest(relayedAddr string, remoteIP string) {
}
// OnRemoteOffer handles an offer from the remote peer and returns true if the message was accepted, false otherwise
// doesn't block, discards the message if connection wasn't ready
func (conn *Conn) OnRemoteOffer(offer OfferAnswer) bool {
client/internal/relay/turn.go
+129
View File
@@ -0,0 +1,129 @@
package relay
import (
"fmt"
"math"
"net"
"sync"
"github.com/pion/logging"
"github.com/pion/stun/v2"
"github.com/pion/turn/v3"
log "github.com/sirupsen/logrus"
)
type PermanentTurn struct {
stunURI *stun.URI
turnURI *stun.URI
stunConn net.PacketConn
turnClient *turn.Client
turnClientListenLock sync.Mutex
relayConn net.PacketConn // represents the remote socket.
srvReflexiveAddress *net.UDPAddr
}
func NewPermanentTurn(stunURL, turnURL *stun.URI) *PermanentTurn {
return &PermanentTurn{
stunURI: stunURL,
turnURI: turnURL,
}
}
func (r *PermanentTurn) Open() error {
stunConn, err := net.ListenPacket("udp4", "0.0.0.0:0")
if err != nil {
return err
}
r.stunConn = stunConn
cfg := &turn.ClientConfig{
STUNServerAddr: toURL(r.stunURI),
TURNServerAddr: toURL(r.turnURI),
Conn: stunConn,
Username: r.turnURI.Username,
Password: r.turnURI.Password,
LoggerFactory: logging.NewDefaultLoggerFactory(),
}
client, err := turn.NewClient(cfg)
if err != nil {
log.Errorf("failed to create turn client: %v", err)
return err
}
r.turnClient = client
r.listen()
relayConn, err := client.Allocate()
if err != nil {
log.Errorf("failed to allocate relay connection: %v", err)
return err
}
r.relayConn = relayConn
srvReflexiveAddress, err := r.discoverPublicIP()
if err != nil {
log.Errorf("failed to discover public IP: %v", err)
return err
}
r.srvReflexiveAddress = srvReflexiveAddress
return nil
}
func (r *PermanentTurn) RelayedAddress() net.Addr {
return r.relayConn.LocalAddr()
}
func (r *PermanentTurn) SrvRefAddr() net.Addr {
return r.srvReflexiveAddress
}
func (r *PermanentTurn) discoverPublicIP() (*net.UDPAddr, error) {
addr, err := r.turnClient.SendBindingRequest()
if err != nil {
log.Errorf("failed to send binding request: %v", err)
return nil, err
}
udpAddr, ok := addr.(*net.UDPAddr)
if !ok {
return nil, fmt.Errorf("failed to cast addr to udp addr")
}
return udpAddr, nil
}
func (r *PermanentTurn) listen() {
if !r.turnClientListenLock.TryLock() {
return
}
go func() {
defer r.turnClientListenLock.Unlock()
buf := make([]byte, math.MaxUint16)
for {
n, from, err := r.stunConn.ReadFrom(buf)
if err != nil {
log.Errorf("Failed to read from stun conn. Exiting loop %v", err)
break
}
_, err = r.turnClient.HandleInbound(buf[:n], from)
if err != nil {
log.Errorf("Failed to handle inbound turn message: %s. Exiting loop", err)
break
}
}
}()
}
func (r *PermanentTurn) PunchHole(mappedAddr net.Addr) error {
_, err := r.relayConn.WriteTo([]byte("Hello"), mappedAddr)
return err
}
func toURL(uri *stun.URI) string {
return fmt.Sprintf("%s:%d", uri.Host, uri.Port)
}
client/internal/relay/turn_test.go
+36
View File
@@ -0,0 +1,36 @@
package relay
import (
"os"
"testing"
"github.com/pion/stun/v2"
"github.com/netbirdio/netbird/util"
)
func TestMain(m *testing.M) {
_ = util.InitLog("trace", "console")
code := m.Run()
os.Exit(code)
}
func TestNewPermanentTurn(t *testing.T) {
turnURI, err := stun.ParseURI("turns:turn.netbird.io:443?transport=tcp")
if err != nil {
t.Errorf("failed to parse stun url: %v", err)
}
turnURI.Username = "1713006060"
turnURI.Password = "pO5Pfx15luZ92mW+FHPa6/LtJ7Y="
stunURI, err := stun.ParseURI("stun:stun.netbird.io:5555")
if err != nil {
t.Errorf("failed to parse stun url: %v", err)
}
turnRelay := NewPermanentTurn(stunURI, turnURI)
err = turnRelay.Open()
if err != nil {
t.Errorf("failed to open turn relay: %v", err)
}
}