Compare commits

..

11 Commits

Author SHA1 Message Date
Jay Wren 7794e93762 Address PR feedback: remove outbound rate limit, improve config docs
Remove rate limiting from StartHandshake (outbound) since DoS
protection only needs to limit inbound handshakes. This also avoids
returning nil from StartHandshake which historically always returned
non-nil. Update config comment to note openssl speed is single-core
and suggest scaling by routines.

Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>
2026-04-10 14:36:32 -04:00
Jay Wren 3df60ae195 Add handshakes.max_rate to limit new handshakes per second
Nebula is vulnerable to DoS via handshake flooding since each incoming
  handshake performs expensive DH operations. This adds a token bucket
  rate limiter to the handshake manager that caps both inbound and
  outbound new handshakes per second. When the limit is reached, new
  handshakes are silently dropped and counted via the
  handshake_manager.rate_limited metric.

  Configured via handshakes.max_rate (default 0 = unlimited).

  Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>
2026-03-30 15:45:38 -04:00
Jack Doan 91d1f4675a properly handle closetunnel packets (#1638) 2026-03-25 11:59:37 -05:00
John Maguire 9f1aef53fa Fix dissector logic (#1626)
* Fix typo in Wireshark dissector

* Fix wireshark dissector prefs_changed logic

The previous logic had several issues:
- Changing only the port number (without toggling all_ports) would
  not re-register the dissector on the new port.
- Turning all_ports off would remove all registrations but only
  re-add the specific port inside a branch that also required
  all_ports to have changed, and never updated default_settings.port.

Simplify to: remove all registrations, then register based on current
prefs, then update the cached state.
2026-03-23 11:15:40 -04:00
Jay R. Wren 1aa1a0476f #ECCN:Open Source in CODEOWNERS (#1632)
gofmt / Run gofmt (push) Failing after 3s
smoke-extra / Run extra smoke tests (push) Failing after 3s
smoke / Run multi node smoke test (push) Failing after 2s
Build and test / Build all and test on ubuntu-linux (push) Failing after 3s
Build and test / Build and test on linux with boringcrypto (push) Failing after 2s
Build and test / Build and test on linux with pkcs11 (push) Failing after 3s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
Salesforce is requesting this in all opensource repositories
2026-03-16 17:07:40 -04:00
Jay R. Wren 7760ccefba fix logging copy pasta (#1621)
gofmt / Run gofmt (push) Failing after 3s
smoke-extra / Run extra smoke tests (push) Failing after 3s
smoke / Run multi node smoke test (push) Failing after 2s
Build and test / Build all and test on ubuntu-linux (push) Failing after 3s
Build and test / Build and test on linux with boringcrypto (push) Failing after 3s
Build and test / Build and test on linux with pkcs11 (push) Failing after 2s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
2026-03-06 14:03:32 -05:00
Jack Doan 51308b845b connection-track ICMP traffic (#1602)
gofmt / Run gofmt (push) Failing after 2s
smoke-extra / Run extra smoke tests (push) Failing after 3s
smoke / Run multi node smoke test (push) Failing after 2s
Build and test / Build all and test on ubuntu-linux (push) Failing after 2s
Build and test / Build and test on linux with boringcrypto (push) Failing after 2s
Build and test / Build and test on linux with pkcs11 (push) Failing after 2s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
* connection-track ICMP and ICMPv6 traffic

* icmpv6 only has identifier on echo
2026-02-18 23:19:37 -06:00
Wade Simmons 422fc2ad1e go fix (#1608)
gofmt / Run gofmt (push) Failing after 3s
smoke-extra / Run extra smoke tests (push) Failing after 2s
smoke / Run multi node smoke test (push) Failing after 2s
Build and test / Build all and test on ubuntu-linux (push) Failing after 2s
Build and test / Build and test on linux with boringcrypto (push) Failing after 3s
Build and test / Build and test on linux with pkcs11 (push) Failing after 2s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
2026-02-17 11:42:14 -05:00
Wade Simmons e8bb874e14 smoke-extra: try AMD-V workaround (#1610)
gofmt / Run gofmt (push) Failing after 3s
smoke-extra / Run extra smoke tests (push) Failing after 2s
smoke / Run multi node smoke test (push) Failing after 2s
Build and test / Build all and test on ubuntu-linux (push) Failing after 3s
Build and test / Build and test on linux with boringcrypto (push) Failing after 2s
Build and test / Build and test on linux with pkcs11 (push) Failing after 2s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
* smoke-extra: try AMD-V workaround

- https://github.com/slackhq/nebula/actions/runs/21995850645/job/63555492676?pr=1602
- https://github.com/actions/runner-images/issues/13202
- https://github.com/cri-o/packaging/pull/306/changes
2026-02-13 12:55:19 -06:00
Jack Doan 353ad1f271 firewall: icmp no longer requires a port spec (#1609) 2026-02-13 11:10:40 -06:00
Jack Doan f573e8a266 Merge commit from fork
gofmt / Run gofmt (push) Failing after 3s
smoke-extra / Run extra smoke tests (push) Failing after 2s
smoke / Run multi node smoke test (push) Failing after 3s
Build and test / Build all and test on ubuntu-linux (push) Failing after 2s
Build and test / Build and test on linux with boringcrypto (push) Failing after 3s
Build and test / Build and test on linux with pkcs11 (push) Failing after 2s
Build and test / Build and test on macos-latest (push) Has been cancelled
Build and test / Build and test on windows-latest (push) Has been cancelled
Newly signed P256 based certificates will have their signature clamped to the low-s form.

Update CHANGELOG.md
2026-02-06 14:26:51 -05:00
63 changed files with 1010 additions and 1515 deletions
+3
View File
@@ -30,6 +30,9 @@ jobs:
- name: add hashicorp source
run: wget -O- https://apt.releases.hashicorp.com/gpg | gpg --dearmor | sudo tee /usr/share/keyrings/hashicorp-archive-keyring.gpg && echo "deb [signed-by=/usr/share/keyrings/hashicorp-archive-keyring.gpg] https://apt.releases.hashicorp.com $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/hashicorp.list
- name: workaround AMD-V issue # https://github.com/cri-o/packaging/pull/306
run: sudo rmmod kvm_amd
- name: install vagrant
run: sudo apt-get update && sudo apt-get install -y vagrant virtualbox
+10 -9
View File
@@ -37,17 +37,18 @@ docker run --name host4 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN -
sleep 1
# grab tcpdump pcaps for debugging
docker exec lighthouse1 tcpdump -i nebula1 -q -w - -U 2>logs/lighthouse1.inside.log >logs/lighthouse1.inside.pcap &
docker exec lighthouse1 tcpdump -i tun0 -q -w - -U 2>logs/lighthouse1.inside.log >logs/lighthouse1.inside.pcap &
docker exec lighthouse1 tcpdump -i eth0 -q -w - -U 2>logs/lighthouse1.outside.log >logs/lighthouse1.outside.pcap &
docker exec host2 tcpdump -i nebula1 -q -w - -U 2>logs/host2.inside.log >logs/host2.inside.pcap &
docker exec host2 tcpdump -i tun0 -q -w - -U 2>logs/host2.inside.log >logs/host2.inside.pcap &
docker exec host2 tcpdump -i eth0 -q -w - -U 2>logs/host2.outside.log >logs/host2.outside.pcap &
docker exec host3 tcpdump -i nebula1 -q -w - -U 2>logs/host3.inside.log >logs/host3.inside.pcap &
docker exec host3 tcpdump -i tun0 -q -w - -U 2>logs/host3.inside.log >logs/host3.inside.pcap &
docker exec host3 tcpdump -i eth0 -q -w - -U 2>logs/host3.outside.log >logs/host3.outside.pcap &
docker exec host4 tcpdump -i nebula1 -q -w - -U 2>logs/host4.inside.log >logs/host4.inside.pcap &
docker exec host4 tcpdump -i tun0 -q -w - -U 2>logs/host4.inside.log >logs/host4.inside.pcap &
docker exec host4 tcpdump -i eth0 -q -w - -U 2>logs/host4.outside.log >logs/host4.outside.pcap &
docker exec host2 ncat -nklv 0.0.0.0 2000 &
docker exec host3 ncat -nklv 0.0.0.0 2000 &
docker exec host4 ncat -nkluv 0.0.0.0 4000 &
docker exec host2 ncat -e '/usr/bin/echo host2' -nkluv 0.0.0.0 3000 &
docker exec host3 ncat -e '/usr/bin/echo host3' -nkluv 0.0.0.0 3000 &
@@ -119,11 +120,11 @@ echo
echo " *** Testing conntrack"
echo
set -x
# host2 can ping host3 now that host3 pinged it first
docker exec host2 ping -c1 192.168.100.3
# host4 can ping host2 once conntrack established
docker exec host2 ping -c1 192.168.100.4
docker exec host4 ping -c1 192.168.100.2
# host2 speaking to host4 on UDP 4000 should allow it to reply, when firewall rules would normally not permit this
docker exec host2 sh -c "/usr/bin/echo host2 | ncat -nuv 192.168.100.4 4000"
docker exec host2 ncat -e '/usr/bin/echo helloagainfromhost2' -nkluv 0.0.0.0 4000 &
docker exec host4 sh -c "/usr/bin/echo host4 | ncat -nuv 192.168.100.2 4000"
docker exec host4 sh -c 'kill 1'
docker exec host3 sh -c 'kill 1'
+15 -1
View File
@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [1.10.3] - 2026-02-06
### Security
- Fix an issue where blocklist bypass is possible when using curve P256 since the signature can have 2 valid representations.
Both fingerprint representations will be tested against the blocklist.
Any newly issued P256 based certificates will have their signature clamped to the low-s form.
Nebula will assert the low-s signature form when validating certificates in a future version. [GHSA-69x3-g4r3-p962](https://github.com/slackhq/nebula/security/advisories/GHSA-69x3-g4r3-p962)
### Changed
- Improve error reporting if nebula fails to start due to a tun device naming issue. (#1588)
## [1.10.2] - 2026-01-21
### Fixed
@@ -775,7 +788,8 @@ created.)
- Initial public release.
[Unreleased]: https://github.com/slackhq/nebula/compare/v1.10.2...HEAD
[Unreleased]: https://github.com/slackhq/nebula/compare/v1.10.3...HEAD
[1.10.3]: https://github.com/slackhq/nebula/releases/tag/v1.10.3
[1.10.2]: https://github.com/slackhq/nebula/releases/tag/v1.10.2
[1.10.1]: https://github.com/slackhq/nebula/releases/tag/v1.10.1
[1.10.0]: https://github.com/slackhq/nebula/releases/tag/v1.10.0
+1
View File
@@ -0,0 +1 @@
#ECCN:Open Source
-1
View File
@@ -1,5 +1,4 @@
//go:build boringcrypto
// +build boringcrypto
package nebula
+18
View File
@@ -141,10 +141,23 @@ func (ncp *CAPool) VerifyCertificate(now time.Time, c Certificate) (*CachedCerti
return nil, err
}
// Pre nebula v1.10.3 could generate signatures in either high or low s form and validation
// of signatures allowed for either. Nebula v1.10.3 and beyond clamps signature generation to low-s form
// but validation still allows for either. Since a change in the signature bytes affects the fingerprint, we
// need to test both forms until such a time comes that we enforce low-s form on signature validation.
fp2, err := CalculateAlternateFingerprint(c)
if err != nil {
return nil, fmt.Errorf("could not calculate alternate fingerprint to verify: %w", err)
}
if fp2 != "" && ncp.IsBlocklisted(fp2) {
return nil, ErrBlockListed
}
cc := CachedCertificate{
Certificate: c,
InvertedGroups: make(map[string]struct{}),
Fingerprint: fp,
fingerprint2: fp2,
signerFingerprint: signer.Fingerprint,
}
@@ -158,6 +171,11 @@ func (ncp *CAPool) VerifyCertificate(now time.Time, c Certificate) (*CachedCerti
// VerifyCachedCertificate is the same as VerifyCertificate other than it operates on a pre-verified structure and
// is a cheaper operation to perform as a result.
func (ncp *CAPool) VerifyCachedCertificate(now time.Time, c *CachedCertificate) error {
// Check any available alternate fingerprint forms for this certificate, re P256 high-s/low-s
if c.fingerprint2 != "" && ncp.IsBlocklisted(c.fingerprint2) {
return ErrBlockListed
}
_, err := ncp.verify(c.Certificate, now, c.Fingerprint, c.signerFingerprint)
return err
}
+43 -4
View File
@@ -5,6 +5,7 @@ import (
"testing"
"time"
"github.com/slackhq/nebula/cert/p256"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -170,6 +171,15 @@ func TestCertificateV1_VerifyP256(t *testing.T) {
_, err = caPool.VerifyCertificate(time.Now(), c)
require.EqualError(t, err, "certificate is in the block list")
// Create a copy of the cert and swap to the alternate form for the signature
nc := c.Copy()
b, err := p256.Swap(c.Signature())
require.NoError(t, err)
require.NoError(t, nc.(*certificateV1).setSignature(b))
_, err = caPool.VerifyCertificate(time.Now(), nc)
require.EqualError(t, err, "certificate is in the block list")
caPool.ResetCertBlocklist()
_, err = caPool.VerifyCertificate(time.Now(), c)
require.NoError(t, err)
@@ -187,7 +197,7 @@ func TestCertificateV1_VerifyP256(t *testing.T) {
require.NoError(t, err)
caPool = NewCAPool()
b, err := caPool.AddCAFromPEM(caPem)
b, err = caPool.AddCAFromPEM(caPem)
require.NoError(t, err)
assert.Empty(t, b)
@@ -196,7 +206,17 @@ func TestCertificateV1_VerifyP256(t *testing.T) {
})
c, _, _, _ = NewTestCert(Version1, Curve_P256, ca, caKey, "test", time.Now(), time.Now().Add(5*time.Minute), nil, nil, []string{"test1"})
_, err = caPool.VerifyCertificate(time.Now(), c)
cc, err := caPool.VerifyCertificate(time.Now(), c)
require.NoError(t, err)
// Reset the blocklist and block the alternate form fingerprint
caPool.ResetCertBlocklist()
caPool.BlocklistFingerprint(cc.fingerprint2)
err = caPool.VerifyCachedCertificate(time.Now(), cc)
require.EqualError(t, err, "certificate is in the block list")
caPool.ResetCertBlocklist()
err = caPool.VerifyCachedCertificate(time.Now(), cc)
require.NoError(t, err)
}
@@ -394,6 +414,15 @@ func TestCertificateV2_VerifyP256(t *testing.T) {
_, err = caPool.VerifyCertificate(time.Now(), c)
require.EqualError(t, err, "certificate is in the block list")
// Create a copy of the cert and swap to the alternate form for the signature
nc := c.Copy()
b, err := p256.Swap(c.Signature())
require.NoError(t, err)
require.NoError(t, nc.(*certificateV2).setSignature(b))
_, err = caPool.VerifyCertificate(time.Now(), nc)
require.EqualError(t, err, "certificate is in the block list")
caPool.ResetCertBlocklist()
_, err = caPool.VerifyCertificate(time.Now(), c)
require.NoError(t, err)
@@ -411,7 +440,7 @@ func TestCertificateV2_VerifyP256(t *testing.T) {
require.NoError(t, err)
caPool = NewCAPool()
b, err := caPool.AddCAFromPEM(caPem)
b, err = caPool.AddCAFromPEM(caPem)
require.NoError(t, err)
assert.Empty(t, b)
@@ -420,7 +449,17 @@ func TestCertificateV2_VerifyP256(t *testing.T) {
})
c, _, _, _ = NewTestCert(Version2, Curve_P256, ca, caKey, "test", time.Now(), time.Now().Add(5*time.Minute), nil, nil, []string{"test1"})
_, err = caPool.VerifyCertificate(time.Now(), c)
cc, err := caPool.VerifyCertificate(time.Now(), c)
require.NoError(t, err)
// Reset the blocklist and block the alternate form fingerprint
caPool.ResetCertBlocklist()
caPool.BlocklistFingerprint(cc.fingerprint2)
err = caPool.VerifyCachedCertificate(time.Now(), cc)
require.EqualError(t, err, "certificate is in the block list")
caPool.ResetCertBlocklist()
err = caPool.VerifyCachedCertificate(time.Now(), cc)
require.NoError(t, err)
}
+33
View File
@@ -4,6 +4,8 @@ import (
"fmt"
"net/netip"
"time"
"github.com/slackhq/nebula/cert/p256"
)
type Version uint8
@@ -110,6 +112,9 @@ type CachedCertificate struct {
InvertedGroups map[string]struct{}
Fingerprint string
signerFingerprint string
// A place to store a 2nd fingerprint if the certificate could have one, such as with P256
fingerprint2 string
}
func (cc *CachedCertificate) String() string {
@@ -152,3 +157,31 @@ func Recombine(v Version, rawCertBytes, publicKey []byte, curve Curve) (Certific
return c, nil
}
// CalculateAlternateFingerprint calculates a 2nd fingerprint representation for P256 certificates
// CAPool blocklist testing through `VerifyCertificate` and `VerifyCachedCertificate` automatically performs this step.
func CalculateAlternateFingerprint(c Certificate) (string, error) {
if c.Curve() != Curve_P256 {
return "", nil
}
nc := c.Copy()
b, err := p256.Swap(nc.Signature())
if err != nil {
return "", err
}
switch v := nc.(type) {
case *certificateV1:
err = v.setSignature(b)
case *certificateV2:
err = v.setSignature(b)
default:
return "", ErrUnknownVersion
}
if err != nil {
return "", err
}
return nc.Fingerprint()
}
+122
View File
@@ -0,0 +1,122 @@
package p256
import (
"crypto/elliptic"
"errors"
"math/big"
"filippo.io/bigmod"
"golang.org/x/crypto/cryptobyte"
"golang.org/x/crypto/cryptobyte/asn1"
)
var halfN = new(big.Int).Rsh(elliptic.P256().Params().N, 1)
var nMod *bigmod.Modulus
func init() {
n, err := bigmod.NewModulus(elliptic.P256().Params().N.Bytes())
if err != nil {
panic(err)
}
nMod = n
}
func IsNormalized(sig []byte) (bool, error) {
r, s, err := parseSignature(sig)
if err != nil {
return false, err
}
return checkLowS(r, s), nil
}
func checkLowS(_, s []byte) bool {
bigS := new(big.Int).SetBytes(s)
// Check if S <= (N/2), because we want to include the midpoint in the set of low-s
return bigS.Cmp(halfN) <= 0
}
func swap(r, s []byte) ([]byte, []byte, error) {
var err error
bigS, err := bigmod.NewNat().SetBytes(s, nMod)
if err != nil {
return nil, nil, err
}
sNormalized := nMod.Nat().Sub(bigS, nMod)
return r, sNormalized.Bytes(nMod), nil
}
func Normalize(sig []byte) ([]byte, error) {
r, s, err := parseSignature(sig)
if err != nil {
return nil, err
}
if checkLowS(r, s) {
return sig, nil
}
newR, newS, err := swap(r, s)
if err != nil {
return nil, err
}
return encodeSignature(newR, newS)
}
// Swap will change sig between its current form to the opposite high or low form.
func Swap(sig []byte) ([]byte, error) {
r, s, err := parseSignature(sig)
if err != nil {
return nil, err
}
newR, newS, err := swap(r, s)
if err != nil {
return nil, err
}
return encodeSignature(newR, newS)
}
// parseSignature taken exactly from crypto/ecdsa/ecdsa.go
func parseSignature(sig []byte) (r, s []byte, err error) {
var inner cryptobyte.String
input := cryptobyte.String(sig)
if !input.ReadASN1(&inner, asn1.SEQUENCE) ||
!input.Empty() ||
!inner.ReadASN1Integer(&r) ||
!inner.ReadASN1Integer(&s) ||
!inner.Empty() {
return nil, nil, errors.New("invalid ASN.1")
}
return r, s, nil
}
func encodeSignature(r, s []byte) ([]byte, error) {
var b cryptobyte.Builder
b.AddASN1(asn1.SEQUENCE, func(b *cryptobyte.Builder) {
addASN1IntBytes(b, r)
addASN1IntBytes(b, s)
})
return b.Bytes()
}
// addASN1IntBytes encodes in ASN.1 a positive integer represented as
// a big-endian byte slice with zero or more leading zeroes.
func addASN1IntBytes(b *cryptobyte.Builder, bytes []byte) {
for len(bytes) > 0 && bytes[0] == 0 {
bytes = bytes[1:]
}
if len(bytes) == 0 {
b.SetError(errors.New("invalid integer"))
return
}
b.AddASN1(asn1.INTEGER, func(c *cryptobyte.Builder) {
if bytes[0]&0x80 != 0 {
c.AddUint8(0)
}
c.AddBytes(bytes)
})
}
+28
View File
@@ -0,0 +1,28 @@
package p256
import (
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"testing"
"github.com/stretchr/testify/require"
)
func TestFlipping(t *testing.T) {
priv, err1 := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err1)
out, err := ecdsa.SignASN1(rand.Reader, priv, []byte("big chungus"))
require.NoError(t, err)
r, s, err := parseSignature(out)
require.NoError(t, err)
r, s1, err := swap(r, s)
require.NoError(t, err)
r, s2, err := swap(r, s1)
require.NoError(t, err)
require.Equal(t, s, s2)
require.NotEqual(t, s, s1)
}
+9
View File
@@ -9,6 +9,8 @@ import (
"fmt"
"net/netip"
"time"
"github.com/slackhq/nebula/cert/p256"
)
// TBSCertificate represents a certificate intended to be signed.
@@ -126,6 +128,13 @@ func (t *TBSCertificate) SignWith(signer Certificate, curve Curve, sp SignerLamb
return nil, err
}
if curve == Curve_P256 {
sig, err = p256.Normalize(sig)
if err != nil {
return nil, err
}
}
err = c.setSignature(sig)
if err != nil {
return nil, err
+46
View File
@@ -9,6 +9,7 @@ import (
"testing"
"time"
"github.com/slackhq/nebula/cert/p256"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -89,3 +90,48 @@ func TestCertificateV1_SignP256(t *testing.T) {
require.NoError(t, err)
assert.NotNil(t, uc)
}
func TestCertificate_SignP256_AlwaysNormalized(t *testing.T) {
before := time.Now().Add(time.Second * -60).Round(time.Second)
after := time.Now().Add(time.Second * 60).Round(time.Second)
pubKey := []byte("01234567890abcedfghij1234567890ab1234567890abcedfghij1234567890ab")
tbs := TBSCertificate{
Version: Version1,
Name: "testing",
Networks: []netip.Prefix{
mustParsePrefixUnmapped("10.1.1.1/24"),
mustParsePrefixUnmapped("10.1.1.2/16"),
},
UnsafeNetworks: []netip.Prefix{
mustParsePrefixUnmapped("9.1.1.2/24"),
mustParsePrefixUnmapped("9.1.1.3/16"),
},
Groups: []string{"test-group1", "test-group2", "test-group3"},
NotBefore: before,
NotAfter: after,
PublicKey: pubKey,
IsCA: true,
Curve: Curve_P256,
}
priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err)
pub := elliptic.Marshal(elliptic.P256(), priv.PublicKey.X, priv.PublicKey.Y)
rawPriv := priv.D.FillBytes(make([]byte, 32))
for i := 0; i < 1000; i++ {
if i&1 == 1 {
tbs.Version = Version1
} else {
tbs.Version = Version2
}
c, err := tbs.Sign(nil, Curve_P256, rawPriv)
require.NoError(t, err)
assert.NotNil(t, c)
assert.True(t, c.CheckSignature(pub))
normie, err := p256.IsNormalized(c.Signature())
require.NoError(t, err)
assert.True(t, normie)
}
}
-4
View File
@@ -1,4 +0,0 @@
package main
const NoSuchFileError = "no such file or directory"
const NoSuchDirError = "no such file or directory"
-4
View File
@@ -1,4 +0,0 @@
package main
const NoSuchFileError = "no such file or directory"
const NoSuchDirError = "no such file or directory"
+4 -5
View File
@@ -11,7 +11,6 @@ import (
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/test"
"github.com/slackhq/nebula/test/device"
"github.com/slackhq/nebula/udp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -53,7 +52,7 @@ func Test_NewConnectionManagerTest(t *testing.T) {
lh := newTestLighthouse()
ifce := &Interface{
hostMap: hostMap,
inside: &device.NoopTun{},
inside: &test.NoopTun{},
outside: &udp.NoopConn{},
firewall: &Firewall{},
lightHouse: lh,
@@ -136,7 +135,7 @@ func Test_NewConnectionManagerTest2(t *testing.T) {
lh := newTestLighthouse()
ifce := &Interface{
hostMap: hostMap,
inside: &device.NoopTun{},
inside: &test.NoopTun{},
outside: &udp.NoopConn{},
firewall: &Firewall{},
lightHouse: lh,
@@ -221,7 +220,7 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) {
lh := newTestLighthouse()
ifce := &Interface{
hostMap: hostMap,
inside: &device.NoopTun{},
inside: &test.NoopTun{},
outside: &udp.NoopConn{},
firewall: &Firewall{},
lightHouse: lh,
@@ -348,7 +347,7 @@ func Test_NewConnectionManagerTest_DisconnectInvalid(t *testing.T) {
lh := newTestLighthouse()
ifce := &Interface{
hostMap: hostMap,
inside: &device.NoopTun{},
inside: &test.NoopTun{},
outside: &udp.NoopConn{},
firewall: &Firewall{},
lightHouse: lh,
-1
View File
@@ -1,5 +1,4 @@
//go:build e2e_testing
// +build e2e_testing
package nebula
+8 -14
View File
@@ -84,30 +84,24 @@ end
function nebula.prefs_changed()
if default_settings.all_ports == nebula.prefs.all_ports and default_settings.port == nebula.prefs.port then
-- Nothing changed, bail
return
end
-- Remove our old dissector
-- Remove all existing registrations
DissectorTable.get("udp.port"):remove_all(nebula)
if nebula.prefs.all_ports and default_settings.all_ports ~= nebula.prefs.all_ports then
default_settings.all_port = nebula.prefs.all_ports
if nebula.prefs.all_ports then
-- Register on every port for hole punch capture
for i=0, 65535 do
DissectorTable.get("udp.port"):add(i, nebula)
end
-- no need to establish again on specific ports
return
else
-- Register on the configured port only
DissectorTable.get("udp.port"):add(nebula.prefs.port, nebula)
end
if default_settings.all_ports ~= nebula.prefs.all_ports then
-- Add our new port dissector
default_settings.port = nebula.prefs.port
DissectorTable.get("udp.port"):add(default_settings.port, nebula)
end
default_settings.all_ports = nebula.prefs.all_ports
default_settings.port = nebula.prefs.port
end
DissectorTable.get("udp.port"):add(default_settings.port, nebula)
+105
View File
@@ -12,6 +12,8 @@ import (
"github.com/slackhq/nebula/cert"
"github.com/slackhq/nebula/cert_test"
"github.com/slackhq/nebula/e2e/router"
"github.com/slackhq/nebula/header"
"github.com/slackhq/nebula/udp"
"github.com/stretchr/testify/assert"
"gopkg.in/yaml.v3"
)
@@ -365,3 +367,106 @@ func TestCrossStackRelaysWork(t *testing.T) {
//theirControl.Stop()
//relayControl.Stop()
}
func TestCloseTunnelAuthenticated(t *testing.T) {
ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "5s"}})
theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "10m"}})
// Share our underlay information
myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
// Start the servers
myControl.Start()
theirControl.Start()
r := router.NewR(t, myControl, theirControl)
r.Log("Assert the tunnel between me and them works")
assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
r.Log("Close the tunnel")
myControl.CloseTunnel(theirVpnIpNet[0].Addr(), false)
r.FlushAll()
waitStart := time.Now()
for {
myIndexes := len(myControl.GetHostmap().Indexes)
theirIndexes := len(theirControl.GetHostmap().Indexes)
if myIndexes == 0 && theirIndexes == 0 {
break
}
since := time.Since(waitStart)
r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
if since > time.Second*6 {
t.Fatal("Tunnel should have been declared inactive after 2 seconds and before 6 seconds")
}
time.Sleep(1 * time.Second)
//r.FlushAll()
}
r.Logf("Happy path success, tunnels were dropped within %v", time.Since(waitStart))
myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
r.Log("Assert another tunnel between me and them works")
assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
if hi == nil {
t.Fatal("There is no hostinfo for this tunnel")
}
myHi := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
if myHi == nil {
t.Fatal("There is no hostinfo for my tunnel")
}
r.Log("It does")
buf := make([]byte, 1024)
hdr := header.H{
Version: 1,
Type: header.CloseTunnel,
Subtype: 0,
Reserved: 0,
RemoteIndex: hi.RemoteIndex,
MessageCounter: 5,
}
out, err := hdr.Encode(buf)
if err != nil {
t.Fatal(err)
}
pkt := &udp.Packet{
To: hi.CurrentRemote,
From: myHi.CurrentRemote,
Data: out,
}
r.InjectUDPPacket(myControl, theirControl, pkt)
r.Log("Injected bogus close tunnel. Let's see!")
waitStart = time.Now()
for {
myIndexes := len(myControl.GetHostmap().Indexes)
theirIndexes := len(theirControl.GetHostmap().Indexes)
if myIndexes == 0 {
t.Fatal("myIndexes should not be 0")
}
if theirIndexes == 0 {
t.Fatal("theirIndexes should not be 0, they should have rejected this bogus packet")
}
since := time.Since(waitStart)
r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
if since > time.Second*4 {
t.Log("The tunnel would have been gone by now")
break
}
time.Sleep(1 * time.Second)
r.FlushAll()
}
myControl.Stop()
theirControl.Stop()
}
+9 -1
View File
@@ -342,6 +342,14 @@ logging:
# after receiving the response for lighthouse queries
#trigger_buffer: 64
# max_rate limits the number of new inbound handshakes per second. Once the limit is reached,
# new handshakes are dropped until the next second. A value of 0 means unlimited (default).
# This is useful for preventing DoS attacks that attempt to exhaust CPU with handshake crypto.
# Running `openssl speed ecdhp256` on your hardware can be a good rule of thumb for choosing
# a max, as each handshake performs similar DH operations. Note that this benchmarks a single
# core, so you may wish to scale the value by the number of `routines` configured.
#max_rate: 0
# Tunnel manager settings
#tunnels:
# drop_inactive controls whether inactive tunnels are maintained or dropped after the inactive_timeout period has
@@ -382,8 +390,8 @@ firewall:
# Rules are comprised of a protocol, port, and one or more of host, group, or CIDR
# Logical evaluation is roughly: port AND proto AND (ca_sha OR ca_name) AND (host OR group OR groups OR cidr) AND (local cidr)
# - port: Takes `0` or `any` as any, a single number `80`, a range `200-901`, or `fragment` to match second and further fragments of fragmented packets (since there is no port available).
# code: same as port but makes more sense when talking about ICMP, TODO: this is not currently implemented in a way that works, use `any`
# proto: `any`, `tcp`, `udp`, or `icmp`
# a port specification is ignored if proto is `icmp`
# host: `any` or a literal hostname, ie `test-host`
# group: `any` or a literal group name, ie `default-group`
# groups: Same as group but accepts a list of values. Multiple values are AND'd together and a certificate would have to contain all groups to pass
+89 -78
View File
@@ -81,14 +81,8 @@ type FirewallConntrack struct {
Conns map[firewall.Packet]*conn
TimerWheel *TimerWheel[firewall.Packet]
// purgeCounter tracks lookups to trigger periodic purge instead of every lookup
purgeCounter uint32
}
// purgeInterval defines how many lookups between purge attempts
const conntrackPurgeInterval = 1024
// FirewallTable is the entry point for a rule, the evaluation order is:
// Proto AND port AND (CA SHA or CA name) AND local CIDR AND (group OR groups OR name OR remote CIDR)
type FirewallTable struct {
@@ -236,7 +230,7 @@ func NewFirewallFromConfig(l *logrus.Logger, cs *CertState, c *config.C) (*Firew
case "drop":
fw.OutSendReject = false
default:
l.WithField("action", inboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
l.WithField("action", outboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
fw.OutSendReject = false
}
@@ -255,20 +249,6 @@ func NewFirewallFromConfig(l *logrus.Logger, cs *CertState, c *config.C) (*Firew
// AddRule properly creates the in memory rule structure for a firewall table.
func (f *Firewall) AddRule(incoming bool, proto uint8, startPort int32, endPort int32, groups []string, host string, cidr, localCidr, caName string, caSha string) error {
// We need this rule string because we generate a hash. Removing this will break firewall reload.
ruleString := fmt.Sprintf(
"incoming: %v, proto: %v, startPort: %v, endPort: %v, groups: %v, host: %v, ip: %v, localIp: %v, caName: %v, caSha: %s",
incoming, proto, startPort, endPort, groups, host, cidr, localCidr, caName, caSha,
)
f.rules += ruleString + "\n"
direction := "incoming"
if !incoming {
direction = "outgoing"
}
f.l.WithField("firewallRule", m{"direction": direction, "proto": proto, "startPort": startPort, "endPort": endPort, "groups": groups, "host": host, "cidr": cidr, "localCidr": localCidr, "caName": caName, "caSha": caSha}).
Info("Firewall rule added")
var (
ft *FirewallTable
fp firewallPort
@@ -286,6 +266,12 @@ func (f *Firewall) AddRule(incoming bool, proto uint8, startPort int32, endPort
case firewall.ProtoUDP:
fp = ft.UDP
case firewall.ProtoICMP, firewall.ProtoICMPv6:
//ICMP traffic doesn't have ports, so we always coerce to "any", even if a value is provided
if startPort != firewall.PortAny {
f.l.WithField("startPort", startPort).Warn("ignoring port specification for ICMP firewall rule")
}
startPort = firewall.PortAny
endPort = firewall.PortAny
fp = ft.ICMP
case firewall.ProtoAny:
fp = ft.AnyProto
@@ -293,6 +279,20 @@ func (f *Firewall) AddRule(incoming bool, proto uint8, startPort int32, endPort
return fmt.Errorf("unknown protocol %v", proto)
}
// We need this rule string because we generate a hash. Removing this will break firewall reload.
ruleString := fmt.Sprintf(
"incoming: %v, proto: %v, startPort: %v, endPort: %v, groups: %v, host: %v, ip: %v, localIp: %v, caName: %v, caSha: %s",
incoming, proto, startPort, endPort, groups, host, cidr, localCidr, caName, caSha,
)
f.rules += ruleString + "\n"
direction := "incoming"
if !incoming {
direction = "outgoing"
}
f.l.WithField("firewallRule", m{"direction": direction, "proto": proto, "startPort": startPort, "endPort": endPort, "groups": groups, "host": host, "cidr": cidr, "localCidr": localCidr, "caName": caName, "caSha": caSha}).
Info("Firewall rule added")
return fp.addRule(f, startPort, endPort, groups, host, cidr, localCidr, caName, caSha)
}
@@ -355,24 +355,31 @@ func AddFirewallRulesFromConfig(l *logrus.Logger, inbound bool, c *config.C, fw
sPort = r.Port
}
startPort, endPort, err := parsePort(sPort)
if err != nil {
return fmt.Errorf("%s rule #%v; %s %s", table, i, errPort, err)
}
var proto uint8
var startPort, endPort int32
switch r.Proto {
case "any":
proto = firewall.ProtoAny
startPort, endPort, err = parsePort(sPort)
case "tcp":
proto = firewall.ProtoTCP
startPort, endPort, err = parsePort(sPort)
case "udp":
proto = firewall.ProtoUDP
startPort, endPort, err = parsePort(sPort)
case "icmp":
proto = firewall.ProtoICMP
startPort = firewall.PortAny
endPort = firewall.PortAny
if sPort != "" {
l.WithField("port", sPort).Warn("ignoring port specification for ICMP firewall rule")
}
default:
return fmt.Errorf("%s rule #%v; proto was not understood; `%s`", table, i, r.Proto)
}
if err != nil {
return fmt.Errorf("%s rule #%v; %s %s", table, i, errPort, err)
}
if r.Cidr != "" && r.Cidr != "any" {
_, err = netip.ParsePrefix(r.Cidr)
@@ -473,7 +480,7 @@ func (f *Firewall) metrics(incoming bool) firewallMetrics {
}
}
// Destroy cleans up any known cyclical references so the object can be free'd my GC. This should be called if a new
// Destroy cleans up any known cyclical references so the object can be freed by GC. This should be called if a new
// firewall object is created
func (f *Firewall) Destroy() {
//TODO: clean references if/when needed
@@ -498,17 +505,14 @@ func (f *Firewall) inConns(fp firewall.Packet, h *HostInfo, caPool *cert.CAPool,
conntrack := f.Conntrack
conntrack.Lock()
// Periodic purge instead of every lookup (major CPU savings)
conntrack.purgeCounter++
if conntrack.purgeCounter >= conntrackPurgeInterval {
conntrack.purgeCounter = 0
ep, has := conntrack.TimerWheel.Purge()
if has {
f.evict(ep)
}
// Purge every time we test
ep, has := conntrack.TimerWheel.Purge()
if has {
f.evict(ep)
}
c, ok := conntrack.Conns[fp]
if !ok {
conntrack.Unlock()
return false
@@ -669,6 +673,13 @@ func (fp firewallPort) match(p firewall.Packet, incoming bool, c *cert.CachedCer
return false
}
// this branch is here to catch traffic from FirewallTable.Any.match and FirewallTable.ICMP.match
if p.Protocol == firewall.ProtoICMP || p.Protocol == firewall.ProtoICMPv6 {
// port numbers are re-used for connection tracking of ICMP,
// but we don't want to actually filter on them.
return fp[firewall.PortAny].match(p, c, caPool)
}
var port int32
if p.Fragment {
@@ -813,10 +824,8 @@ func (fr *FirewallRule) isAny(groups []string, host string, cidr string) bool {
return true
}
for _, group := range groups {
if group == "any" {
return true
}
if slices.Contains(groups, "any") {
return true
}
if host == "any" {
@@ -1027,54 +1036,56 @@ func (r *rule) sanity() error {
}
}
if r.Code != "" {
return fmt.Errorf("code specified as [%s]. Support for 'code' will be dropped in a future release, as it has never been functional", r.Code)
}
//todo alert on cidr-any
return nil
}
func parsePort(s string) (startPort, endPort int32, err error) {
func parsePort(s string) (int32, int32, error) {
var err error
const notAPort int32 = -2
if s == "any" {
startPort = firewall.PortAny
endPort = firewall.PortAny
} else if s == "fragment" {
startPort = firewall.PortFragment
endPort = firewall.PortFragment
} else if strings.Contains(s, `-`) {
sPorts := strings.SplitN(s, `-`, 2)
sPorts[0] = strings.Trim(sPorts[0], " ")
sPorts[1] = strings.Trim(sPorts[1], " ")
if len(sPorts) != 2 || sPorts[0] == "" || sPorts[1] == "" {
return 0, 0, fmt.Errorf("appears to be a range but could not be parsed; `%s`", s)
}
rStartPort, err := strconv.Atoi(sPorts[0])
if err != nil {
return 0, 0, fmt.Errorf("beginning range was not a number; `%s`", sPorts[0])
}
rEndPort, err := strconv.Atoi(sPorts[1])
if err != nil {
return 0, 0, fmt.Errorf("ending range was not a number; `%s`", sPorts[1])
}
startPort = int32(rStartPort)
endPort = int32(rEndPort)
if startPort == firewall.PortAny {
endPort = firewall.PortAny
}
} else {
return firewall.PortAny, firewall.PortAny, nil
}
if s == "fragment" {
return firewall.PortFragment, firewall.PortFragment, nil
}
if !strings.Contains(s, `-`) {
rPort, err := strconv.Atoi(s)
if err != nil {
return 0, 0, fmt.Errorf("was not a number; `%s`", s)
return notAPort, notAPort, fmt.Errorf("was not a number; `%s`", s)
}
startPort = int32(rPort)
endPort = startPort
return int32(rPort), int32(rPort), nil
}
return
sPorts := strings.SplitN(s, `-`, 2)
for i := range sPorts {
sPorts[i] = strings.Trim(sPorts[i], " ")
}
if len(sPorts) != 2 || sPorts[0] == "" || sPorts[1] == "" {
return notAPort, notAPort, fmt.Errorf("appears to be a range but could not be parsed; `%s`", s)
}
rStartPort, err := strconv.Atoi(sPorts[0])
if err != nil {
return notAPort, notAPort, fmt.Errorf("beginning range was not a number; `%s`", sPorts[0])
}
rEndPort, err := strconv.Atoi(sPorts[1])
if err != nil {
return notAPort, notAPort, fmt.Errorf("ending range was not a number; `%s`", sPorts[1])
}
startPort := int32(rStartPort)
endPort := int32(rEndPort)
if startPort == firewall.PortAny {
endPort = firewall.PortAny
}
return startPort, endPort, nil
}
+6 -1
View File
@@ -22,7 +22,10 @@ const (
type Packet struct {
LocalAddr netip.Addr
RemoteAddr netip.Addr
LocalPort uint16
// LocalPort is the destination port for incoming traffic, or the source port for outgoing. Zero for ICMP.
LocalPort uint16
// RemotePort is the source port for incoming traffic, or the destination port for outgoing.
// For ICMP, it's the "identifier". This is only used for connection tracking, actual firewall rules will not filter on ICMP identifier
RemotePort uint16
Protocol uint8
Fragment bool
@@ -46,6 +49,8 @@ func (fp Packet) MarshalJSON() ([]byte, error) {
proto = "tcp"
case ProtoICMP:
proto = "icmp"
case ProtoICMPv6:
proto = "icmpv6"
case ProtoUDP:
proto = "udp"
default:
+158 -6
View File
@@ -87,9 +87,10 @@ func TestFirewall_AddRule(t *testing.T) {
fw = NewFirewall(l, time.Second, time.Minute, time.Hour, c)
require.NoError(t, fw.AddRule(true, firewall.ProtoICMP, 1, 1, []string{}, "h1", "", "", "", ""))
assert.Nil(t, fw.InRules.ICMP[1].Any.Any)
assert.Empty(t, fw.InRules.ICMP[1].Any.Groups)
assert.Contains(t, fw.InRules.ICMP[1].Any.Hosts, "h1")
//no matter what port is given for icmp, it should end up as "any"
assert.Nil(t, fw.InRules.ICMP[firewall.PortAny].Any.Any)
assert.Empty(t, fw.InRules.ICMP[firewall.PortAny].Any.Groups)
assert.Contains(t, fw.InRules.ICMP[firewall.PortAny].Any.Hosts, "h1")
fw = NewFirewall(l, time.Second, time.Minute, time.Hour, c)
require.NoError(t, fw.AddRule(false, firewall.ProtoAny, 1, 1, []string{}, "", ti.String(), "", "", ""))
@@ -734,6 +735,150 @@ func TestFirewall_DropConntrackReload(t *testing.T) {
assert.Equal(t, fw.Drop(p, false, &h, cp, nil), ErrNoMatchingRule)
}
func TestFirewall_ICMPPortBehavior(t *testing.T) {
l := test.NewLogger()
ob := &bytes.Buffer{}
l.SetOutput(ob)
myVpnNetworksTable := new(bart.Lite)
myVpnNetworksTable.Insert(netip.MustParsePrefix("1.1.1.1/8"))
network := netip.MustParsePrefix("1.2.3.4/24")
c := cert.CachedCertificate{
Certificate: &dummyCert{
name: "host1",
networks: []netip.Prefix{network},
groups: []string{"default-group"},
issuer: "signer-shasum",
},
InvertedGroups: map[string]struct{}{"default-group": {}},
}
h := HostInfo{
ConnectionState: &ConnectionState{
peerCert: &c,
},
vpnAddrs: []netip.Addr{network.Addr()},
}
h.buildNetworks(myVpnNetworksTable, c.Certificate)
cp := cert.NewCAPool()
templ := firewall.Packet{
LocalAddr: netip.MustParseAddr("1.2.3.4"),
RemoteAddr: netip.MustParseAddr("1.2.3.4"),
Protocol: firewall.ProtoICMP,
Fragment: false,
}
t.Run("ICMP allowed", func(t *testing.T) {
fw := NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate)
require.NoError(t, fw.AddRule(true, firewall.ProtoICMP, 0, 0, []string{"any"}, "", "", "", "", ""))
t.Run("zero ports", func(t *testing.T) {
p := templ.Copy()
p.LocalPort = 0
p.RemotePort = 0
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
require.NoError(t, fw.Drop(*p, true, &h, cp, nil))
//now also allow outbound
require.NoError(t, fw.Drop(*p, false, &h, cp, nil))
})
t.Run("nonzero ports", func(t *testing.T) {
p := templ.Copy()
p.LocalPort = 0xabcd
p.RemotePort = 0x1234
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
require.NoError(t, fw.Drop(*p, true, &h, cp, nil))
//now also allow outbound
require.NoError(t, fw.Drop(*p, false, &h, cp, nil))
})
})
t.Run("Any proto, some ports allowed", func(t *testing.T) {
fw := NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate)
require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 80, 444, []string{"any"}, "", "", "", "", ""))
t.Run("zero ports, still blocked", func(t *testing.T) {
p := templ.Copy()
p.LocalPort = 0
p.RemotePort = 0
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule)
//now also allow outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
})
t.Run("nonzero ports, still blocked", func(t *testing.T) {
p := templ.Copy()
p.LocalPort = 0xabcd
p.RemotePort = 0x1234
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule)
//now also allow outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
})
t.Run("nonzero, matching ports, still blocked", func(t *testing.T) {
p := templ.Copy()
p.LocalPort = 80
p.RemotePort = 80
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
assert.Equal(t, fw.Drop(*p, true, &h, cp, nil), ErrNoMatchingRule)
//now also allow outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
})
})
t.Run("Any proto, any port", func(t *testing.T) {
fw := NewFirewall(l, time.Second, time.Minute, time.Hour, c.Certificate)
require.NoError(t, fw.AddRule(true, firewall.ProtoAny, 0, 0, []string{"any"}, "", "", "", "", ""))
t.Run("zero ports, allowed", func(t *testing.T) {
resetConntrack(fw)
p := templ.Copy()
p.LocalPort = 0
p.RemotePort = 0
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
require.NoError(t, fw.Drop(*p, true, &h, cp, nil))
//now also allow outbound
require.NoError(t, fw.Drop(*p, false, &h, cp, nil))
})
t.Run("nonzero ports, allowed", func(t *testing.T) {
resetConntrack(fw)
p := templ.Copy()
p.LocalPort = 0xabcd
p.RemotePort = 0x1234
// Drop outbound
assert.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
// Allow inbound
resetConntrack(fw)
require.NoError(t, fw.Drop(*p, true, &h, cp, nil))
//now also allow outbound
require.NoError(t, fw.Drop(*p, false, &h, cp, nil))
//different ID is blocked
p.RemotePort++
require.Equal(t, fw.Drop(*p, false, &h, cp, nil), ErrNoMatchingRule)
})
})
}
func TestFirewall_DropIPSpoofing(t *testing.T) {
l := test.NewLogger()
ob := &bytes.Buffer{}
@@ -919,11 +1064,11 @@ func TestNewFirewallFromConfig(t *testing.T) {
// Test code/port error
conf = config.NewC(l)
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"code": "a", "host": "testh"}}}
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"code": "a", "host": "testh", "proto": "any"}}}
_, err = NewFirewallFromConfig(l, cs, conf)
require.EqualError(t, err, "firewall.outbound rule #0; code was not a number; `a`")
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"port": "a", "host": "testh"}}}
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"port": "a", "host": "testh", "proto": "any"}}}
_, err = NewFirewallFromConfig(l, cs, conf)
require.EqualError(t, err, "firewall.outbound rule #0; port was not a number; `a`")
@@ -973,7 +1118,14 @@ func TestAddFirewallRulesFromConfig(t *testing.T) {
mf = &mockFirewall{}
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"port": "1", "proto": "icmp", "host": "a"}}}
require.NoError(t, AddFirewallRulesFromConfig(l, false, conf, mf))
assert.Equal(t, addRuleCall{incoming: false, proto: firewall.ProtoICMP, startPort: 1, endPort: 1, groups: nil, host: "a", ip: "", localIp: ""}, mf.lastCall)
assert.Equal(t, addRuleCall{incoming: false, proto: firewall.ProtoICMP, startPort: firewall.PortAny, endPort: firewall.PortAny, groups: nil, host: "a", ip: "", localIp: ""}, mf.lastCall)
// Test adding icmp rule no port
conf = config.NewC(l)
mf = &mockFirewall{}
conf.Settings["firewall"] = map[string]any{"outbound": []any{map[string]any{"proto": "icmp", "host": "a"}}}
require.NoError(t, AddFirewallRulesFromConfig(l, false, conf, mf))
assert.Equal(t, addRuleCall{incoming: false, proto: firewall.ProtoICMP, startPort: firewall.PortAny, endPort: firewall.PortAny, groups: nil, host: "a", ip: "", localIp: ""}, mf.lastCall)
// Test adding any rule
conf = config.NewC(l)
+4 -3
View File
@@ -4,6 +4,7 @@ go 1.25
require (
dario.cat/mergo v1.0.2
filippo.io/bigmod v0.1.0
github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be
github.com/armon/go-radix v1.0.0
github.com/cyberdelia/go-metrics-graphite v0.0.0-20161219230853-39f87cc3b432
@@ -30,11 +31,11 @@ require (
golang.org/x/sys v0.40.0
golang.org/x/term v0.39.0
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2
golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb
golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b
golang.zx2c4.com/wireguard/windows v0.5.3
google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1
gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c
gvisor.dev/gvisor v0.0.0-20240423190808-9d7a357edefe
)
require (
@@ -50,6 +51,6 @@ require (
github.com/vishvananda/netns v0.0.5 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
golang.org/x/mod v0.31.0 // indirect
golang.org/x/time v0.7.0 // indirect
golang.org/x/time v0.5.0 // indirect
golang.org/x/tools v0.40.0 // indirect
)
+8 -6
View File
@@ -1,6 +1,8 @@
cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
filippo.io/bigmod v0.1.0 h1:UNzDk7y9ADKST+axd9skUpBQeW7fG2KrTZyOE4uGQy8=
filippo.io/bigmod v0.1.0/go.mod h1:OjOXDNlClLblvXdwgFFOQFJEocLhhtai8vGLy0JCZlI=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -216,8 +218,8 @@ golang.org/x/term v0.39.0/go.mod h1:yxzUCTP/U+FzoxfdKmLaA0RV1WgE0VY7hXBwKtY/4ww=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
@@ -231,8 +233,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 h1:B82qJJgjvYKsXS9jeunTOisW56dUokqW/FOteYJJ/yg=
golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2/go.mod h1:deeaetjYA+DHMHg+sMSMI58GrEteJUUzzw7en6TJQcI=
golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb h1:whnFRlWMcXI9d+ZbWg+4sHnLp52d5yiIPUxMBSt4X9A=
golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb/go.mod h1:rpwXGsirqLqN2L0JDJQlwOboGHmptD5ZD6T2VmcqhTw=
golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b h1:J1CaxgLerRR5lgx3wnr6L04cJFbWoceSK9JWBdglINo=
golang.zx2c4.com/wireguard v0.0.0-20230325221338-052af4a8072b/go.mod h1:tqur9LnfstdR9ep2LaJT4lFUl0EjlHtge+gAjmsHUG4=
golang.zx2c4.com/wireguard/windows v0.5.3 h1:On6j2Rpn3OEMXqBq00QEDC7bWSZrPIHKIus8eIuExIE=
golang.zx2c4.com/wireguard/windows v0.5.3/go.mod h1:9TEe8TJmtwyQebdFwAkEWOPr3prrtqm+REGFifP60hI=
google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -258,5 +260,5 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c h1:m/r7OM+Y2Ty1sgBQ7Qb27VgIMBW8ZZhT4gLnUyDIhzI=
gvisor.dev/gvisor v0.0.0-20250503011706-39ed1f5ac29c/go.mod h1:3r5CMtNQMKIvBlrmM9xWUNamjKBYPOWyXOjmg5Kts3g=
gvisor.dev/gvisor v0.0.0-20240423190808-9d7a357edefe h1:fre4i6mv4iBuz5lCMOzHD1rH1ljqHWSICFmZRbbgp3g=
gvisor.dev/gvisor v0.0.0-20240423190808-9d7a357edefe/go.mod h1:sxc3Uvk/vHcd3tj7/DHVBoR5wvWT/MmRq2pj7HRJnwU=
+58 -9
View File
@@ -23,22 +23,25 @@ const (
DefaultHandshakeRetries = 10
DefaultHandshakeTriggerBuffer = 64
DefaultUseRelays = true
DefaultMaxHandshakeRate = 0 // 0 means unlimited
)
var (
defaultHandshakeConfig = HandshakeConfig{
tryInterval: DefaultHandshakeTryInterval,
retries: DefaultHandshakeRetries,
triggerBuffer: DefaultHandshakeTriggerBuffer,
useRelays: DefaultUseRelays,
tryInterval: DefaultHandshakeTryInterval,
retries: DefaultHandshakeRetries,
triggerBuffer: DefaultHandshakeTriggerBuffer,
useRelays: DefaultUseRelays,
maxHandshakeRate: DefaultMaxHandshakeRate,
}
)
type HandshakeConfig struct {
tryInterval time.Duration
retries int64
triggerBuffer int
useRelays bool
tryInterval time.Duration
retries int64
triggerBuffer int
useRelays bool
maxHandshakeRate int
messageMetrics *MessageMetrics
}
@@ -58,9 +61,15 @@ type HandshakeManager struct {
messageMetrics *MessageMetrics
metricInitiated metrics.Counter
metricTimedOut metrics.Counter
metricRateLimited metrics.Counter
f *Interface
l *logrus.Logger
// Rate limiting for new handshakes (token bucket)
rateBucket int // tokens currently available
rateMax int // max tokens (== max handshakes per second), 0 means unlimited
rateLastTick time.Time
// can be used to trigger outbound handshake for the given vpnIp
trigger chan netip.Addr
}
@@ -116,10 +125,41 @@ func NewHandshakeManager(l *logrus.Logger, mainHostMap *HostMap, lightHouse *Lig
messageMetrics: config.messageMetrics,
metricInitiated: metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
metricTimedOut: metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
metricRateLimited: metrics.GetOrRegisterCounter("handshake_manager.rate_limited", nil),
rateBucket: config.maxHandshakeRate,
rateMax: config.maxHandshakeRate,
rateLastTick: time.Now(),
l: l,
}
}
// handshakeRateAllow checks the token bucket rate limiter and returns true if a
// new handshake is allowed. Must be called with hm.Lock held.
func (hm *HandshakeManager) handshakeRateAllow(now time.Time) bool {
if hm.rateMax == 0 {
return true
}
// Refill tokens based on elapsed time
elapsed := now.Sub(hm.rateLastTick)
if elapsed >= time.Second {
// Add tokens for full seconds elapsed
tokens := int(elapsed/time.Second) * hm.rateMax
hm.rateBucket += tokens
if hm.rateBucket > hm.rateMax {
hm.rateBucket = hm.rateMax
}
hm.rateLastTick = now
}
if hm.rateBucket > 0 {
hm.rateBucket--
return true
}
return false
}
func (hm *HandshakeManager) Run(ctx context.Context) {
clockSource := time.NewTicker(hm.config.tryInterval)
defer clockSource.Stop()
@@ -149,6 +189,15 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head
case header.HandshakeIXPSK0:
switch h.MessageCounter {
case 1:
// Check rate limit for new incoming handshakes
hm.Lock()
allowed := hm.handshakeRateAllow(time.Now())
hm.Unlock()
if !allowed {
hm.metricRateLimited.Inc(1)
hm.l.WithField("from", via).Debug("Handshake rate limit reached, dropping incoming handshake")
return
}
ixHandshakeStage1(hm.f, via, packet, h)
case 2:
@@ -590,7 +639,7 @@ func (hm *HandshakeManager) allocateIndex(hh *HandshakeHostInfo) error {
hm.Lock()
defer hm.Unlock()
for i := 0; i < 32; i++ {
for range 32 {
index, err := generateIndex(hm.l)
if err != nil {
return err
+62
View File
@@ -65,6 +65,68 @@ func Test_NewHandshakeManagerVpnIp(t *testing.T) {
assert.NotContains(t, blah.vpnIps, ip)
}
func Test_HandshakeManagerRateLimit(t *testing.T) {
l := test.NewLogger()
localrange := netip.MustParsePrefix("10.1.1.1/24")
preferredRanges := []netip.Prefix{localrange}
mainHM := newHostMap(l)
mainHM.preferredRanges.Store(&preferredRanges)
lh := newTestLighthouse()
config := defaultHandshakeConfig
config.maxHandshakeRate = 2
hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, config)
hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
now := time.Now()
// Should allow up to maxHandshakeRate handshakes
hm.Lock()
assert.True(t, hm.handshakeRateAllow(now), "first handshake should be allowed")
assert.True(t, hm.handshakeRateAllow(now), "second handshake should be allowed")
assert.False(t, hm.handshakeRateAllow(now), "third handshake should be rate limited")
hm.Unlock()
// After advancing time by 1 second, tokens should refill
hm.Lock()
assert.True(t, hm.handshakeRateAllow(now.Add(time.Second)), "handshake should be allowed after token refill")
hm.Unlock()
}
func Test_HandshakeManagerRateLimitUnlimited(t *testing.T) {
l := test.NewLogger()
localrange := netip.MustParsePrefix("10.1.1.1/24")
preferredRanges := []netip.Prefix{localrange}
mainHM := newHostMap(l)
mainHM.preferredRanges.Store(&preferredRanges)
lh := newTestLighthouse()
cs := &CertState{
initiatingVersion: cert.Version1,
privateKey: []byte{},
v1Cert: &dummyCert{version: cert.Version1},
v1HandshakeBytes: []byte{},
}
// Default config has maxHandshakeRate=0 (unlimited)
hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, defaultHandshakeConfig)
hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
hm.f.pki.cs.Store(cs)
// Should allow many handshakes with no limit
// Limited to 10 due to test lighthouse query channel buffer
for i := 0; i < 10; i++ {
ip := netip.MustParseAddr("172.1.1.1").As16()
ip[15] = byte(i + 1)
addr := netip.AddrFrom16(ip)
h := hm.StartHandshake(addr, nil)
assert.NotNil(t, h, "handshake %d should be allowed with unlimited rate", i)
}
}
func testCountTimerWheelEntries(tw *LockingTimerWheel[netip.Addr]) (c int) {
for _, i := range tw.t.wheel {
n := i.Head
-1
View File
@@ -1,5 +1,4 @@
//go:build e2e_testing
// +build e2e_testing
package nebula
-254
View File
@@ -2,8 +2,6 @@ package nebula
import (
"net/netip"
"sync"
"time"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/firewall"
@@ -13,258 +11,6 @@ import (
"github.com/slackhq/nebula/routing"
)
// preEncryptionPacket holds packet data before batch encryption
type preEncryptionPacket struct {
hostinfo *HostInfo
ci *ConnectionState
packet []byte
out []byte
}
// Pool for preEncryptionBatch slices to reduce allocations
var preEncryptionBatchPool = sync.Pool{
New: func() any {
// Pre-allocate with reasonable capacity
batch := make([]preEncryptionPacket, 0, 128)
return &batch
},
}
// consumeInsidePackets processes multiple packets in a batch for improved performance
// packets: slice of packet buffers to process
// sizes: slice of packet sizes
// count: number of packets to process
// outs: slice of output buffers (one per packet) with virtio headroom
// q: queue index
// localCache: firewall conntrack cache
// batchPackets: pre-allocated slice for accumulating encrypted packets
// batchAddrs: pre-allocated slice for accumulating destination addresses
func (f *Interface) consumeInsidePackets(packets [][]byte, sizes []int, count int, outs [][]byte, nb []byte, q int, localCache firewall.ConntrackCache, batchPackets *[][]byte, batchAddrs *[]netip.AddrPort) {
// Reusable per-packet state
fwPacket := &firewall.Packet{}
// Reset batch accumulation slices (reuse capacity)
*batchPackets = (*batchPackets)[:0]
*batchAddrs = (*batchAddrs)[:0]
// Collect packets for batched encryption
preEncryptionBatch := make([]preEncryptionPacket, 0, count)
// Process each packet in the batch
for i := 0; i < count; i++ {
packet := packets[i][:sizes[i]]
out := outs[i]
// Inline the consumeInsidePacket logic for better performance
err := newPacket(packet, false, fwPacket)
if err != nil {
if f.l.Level >= logrus.DebugLevel {
f.l.WithField("packet", packet).Debugf("Error while validating outbound packet: %s", err)
}
continue
}
// Ignore local broadcast packets
if f.dropLocalBroadcast {
if f.myBroadcastAddrsTable.Contains(fwPacket.RemoteAddr) {
continue
}
}
if f.myVpnAddrsTable.Contains(fwPacket.RemoteAddr) {
// Immediately forward packets from self to self.
if immediatelyForwardToSelf {
_, err := f.readers[q].Write(packet)
if err != nil {
f.l.WithError(err).Error("Failed to forward to tun")
}
}
continue
}
// Ignore multicast packets
if f.dropMulticast && fwPacket.RemoteAddr.IsMulticast() {
continue
}
hostinfo, ready := f.getOrHandshakeConsiderRouting(fwPacket, func(hh *HandshakeHostInfo) {
hh.cachePacket(f.l, header.Message, 0, packet, f.sendMessageNow, f.cachedPacketMetrics)
})
if hostinfo == nil {
f.rejectInside(packet, out, q)
if f.l.Level >= logrus.DebugLevel {
f.l.WithField("vpnAddr", fwPacket.RemoteAddr).
WithField("fwPacket", fwPacket).
Debugln("dropping outbound packet, vpnAddr not in our vpn networks or in unsafe networks")
}
continue
}
if !ready {
continue
}
dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache)
if dropReason != nil {
f.rejectInside(packet, out, q)
if f.l.Level >= logrus.DebugLevel {
hostinfo.logger(f.l).
WithField("fwPacket", fwPacket).
WithField("reason", dropReason).
Debugln("dropping outbound packet")
}
continue
}
// Prepare packet for batch encryption
ci := hostinfo.ConnectionState
if ci.eKey == nil {
continue
}
// Check if this needs relay - if so, send immediately and skip batching
useRelay := !hostinfo.remote.IsValid()
if useRelay {
// Handle relay sends individually (less common path)
f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, packet, nb, out, q)
continue
}
// Collect for batched encryption
preEncryptionBatch = append(preEncryptionBatch, preEncryptionPacket{
hostinfo: hostinfo,
ci: ci,
packet: packet,
out: out,
})
}
// BATCH ENCRYPTION: Process all collected packets
if len(preEncryptionBatch) > 0 {
f.encryptBatch(preEncryptionBatch, nb, batchPackets, batchAddrs)
}
// Send all accumulated packets in one batch
if len(*batchPackets) > 0 {
batchSize := len(*batchPackets)
n, err := f.writers[q].WriteMulti(*batchPackets, *batchAddrs)
if err != nil {
f.l.WithError(err).WithField("sent", n).WithField("total", batchSize).Error("Failed to send batch")
}
}
}
// encryptBatch processes multiple packets, grouping by ConnectionState to reduce lock acquisitions
func (f *Interface) encryptBatch(batch []preEncryptionPacket, nb []byte, batchPackets *[][]byte, batchAddrs *[]netip.AddrPort) {
lockStart := time.Now()
lockAcquisitions := int64(0)
if noiseutil.EncryptLockNeeded {
// Group packets by ConnectionState to minimize lock acquisitions
// Process packets in order but batch lock acquisitions per CI
var currentCI *ConnectionState
var lockHeld bool
for i := range batch {
ci := batch[i].ci
hostinfo := batch[i].hostinfo
// Validate packet data to prevent nil pointer dereference
if ci == nil || hostinfo == nil {
continue
}
// Switch locks if we're moving to a different ConnectionState
if ci != currentCI {
if lockHeld {
currentCI.writeLock.Unlock()
}
ci.writeLock.Lock()
lockAcquisitions++
currentCI = ci
lockHeld = true
}
c := ci.messageCounter.Add(1)
out := header.Encode(batch[i].out, header.Version, header.Message, 0, hostinfo.remoteIndexId, c)
f.connectionManager.Out(hostinfo)
// Query lighthouse if needed
if hostinfo.lastRebindCount != f.rebindCount {
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
hostinfo.lastRebindCount = f.rebindCount
if f.l.Level >= logrus.DebugLevel {
f.l.WithField("vpnAddrs", hostinfo.vpnAddrs).Debug("Lighthouse update triggered for punch due to rebind counter")
}
}
var err error
out, err = ci.eKey.EncryptDanger(out, out, batch[i].packet, c, nb)
if err != nil {
hostinfo.logger(f.l).WithError(err).
WithField("counter", c).
Error("Failed to encrypt outgoing packet")
continue
}
// Add to output batches
*batchPackets = append(*batchPackets, out)
*batchAddrs = append(*batchAddrs, hostinfo.remote)
}
// Release final lock
if lockHeld {
currentCI.writeLock.Unlock()
}
} else {
// No locks needed - process directly
for i := range batch {
ci := batch[i].ci
hostinfo := batch[i].hostinfo
// Validate packet data to prevent nil pointer dereference
if ci == nil || hostinfo == nil {
continue
}
c := ci.messageCounter.Add(1)
out := header.Encode(batch[i].out, header.Version, header.Message, 0, hostinfo.remoteIndexId, c)
f.connectionManager.Out(hostinfo)
// Query lighthouse if needed
if hostinfo.lastRebindCount != f.rebindCount {
f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
hostinfo.lastRebindCount = f.rebindCount
if f.l.Level >= logrus.DebugLevel {
f.l.WithField("vpnAddrs", hostinfo.vpnAddrs).Debug("Lighthouse update triggered for punch due to rebind counter")
}
}
var err error
out, err = ci.eKey.EncryptDanger(out, out, batch[i].packet, c, nb)
if err != nil {
hostinfo.logger(f.l).WithError(err).
WithField("counter", c).
Error("Failed to encrypt outgoing packet")
continue
}
// Add to output batches
*batchPackets = append(*batchPackets, out)
*batchAddrs = append(*batchAddrs, hostinfo.remote)
}
}
// Record metrics
encryptionTime := time.Since(lockStart)
if noiseutil.EncryptLockNeeded {
f.batchMetrics.lockAcquisitions.Inc(lockAcquisitions)
}
f.batchMetrics.encryptionTime.Update(encryptionTime.Nanoseconds())
f.batchMetrics.batchSize.Update(int64(len(batch)))
}
func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb, out []byte, q int, localCache firewall.ConntrackCache) {
err := newPacket(packet, false, fwPacket)
if err != nil {
-1
View File
@@ -1,5 +1,4 @@
//go:build darwin || dragonfly || freebsd || netbsd || openbsd
// +build darwin dragonfly freebsd netbsd openbsd
package nebula
-1
View File
@@ -1,5 +1,4 @@
//go:build !darwin && !dragonfly && !freebsd && !netbsd && !openbsd
// +build !darwin,!dragonfly,!freebsd,!netbsd,!openbsd
package nebula
+16 -59
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"io"
"net/netip"
"os"
"runtime"
@@ -21,7 +22,6 @@ import (
)
const mtu = 9001
const virtioNetHdrLen = overlay.VirtioNetHdrLen
type InterfaceConfig struct {
HostMap *HostMap
@@ -50,13 +50,6 @@ type InterfaceConfig struct {
l *logrus.Logger
}
type batchMetrics struct {
udpReadSize metrics.Histogram
encryptionTime metrics.Histogram // Time spent in encryption (including lock waits)
batchSize metrics.Histogram // Dynamic batch sizes being used
lockAcquisitions metrics.Counter // Number of lock acquisitions (should be minimal)
}
type Interface struct {
hostMap *HostMap
outside udp.Conn
@@ -94,12 +87,11 @@ type Interface struct {
conntrackCacheTimeout time.Duration
writers []udp.Conn
readers []overlay.BatchReadWriter
readers []io.ReadWriteCloser
metricHandshakes metrics.Histogram
messageMetrics *MessageMetrics
cachedPacketMetrics *cachedPacketMetrics
batchMetrics *batchMetrics
l *logrus.Logger
}
@@ -186,7 +178,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
routines: c.routines,
version: c.version,
writers: make([]udp.Conn, c.routines),
readers: make([]overlay.BatchReadWriter, c.routines),
readers: make([]io.ReadWriteCloser, c.routines),
myVpnNetworks: cs.myVpnNetworks,
myVpnNetworksTable: cs.myVpnNetworksTable,
myVpnAddrs: cs.myVpnAddrs,
@@ -202,12 +194,6 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
sent: metrics.GetOrRegisterCounter("hostinfo.cached_packets.sent", nil),
dropped: metrics.GetOrRegisterCounter("hostinfo.cached_packets.dropped", nil),
},
batchMetrics: &batchMetrics{
udpReadSize: metrics.GetOrRegisterHistogram("batch.udp_read_size", nil, metrics.NewUniformSample(1024)),
encryptionTime: metrics.GetOrRegisterHistogram("batch.encryption_time_ns", nil, metrics.NewUniformSample(1024)),
batchSize: metrics.GetOrRegisterHistogram("batch.size", nil, metrics.NewUniformSample(1024)),
lockAcquisitions: metrics.GetOrRegisterCounter("batch.lock_acquisitions", nil),
},
l: c.l,
}
@@ -247,7 +233,7 @@ func (f *Interface) activate() {
metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
// Prepare n tun queues
var reader overlay.BatchReadWriter = f.inside
var reader io.ReadWriteCloser = f.inside
for i := 0; i < f.routines; i++ {
if i > 0 {
reader, err = f.inside.NewMultiQueueReader()
@@ -288,68 +274,39 @@ func (f *Interface) listenOut(i int) {
ctCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
lhh := f.lightHouse.NewRequestHandler()
// Pre-allocate output buffers for batch processing
batchSize := li.BatchSize()
outs := make([][]byte, batchSize)
for idx := range outs {
// Allocate full buffer with virtio header space
outs[idx] = make([]byte, virtioNetHdrLen, virtioNetHdrLen+udp.MTU)
}
plaintext := make([]byte, udp.MTU)
h := &header.H{}
fwPacket := &firewall.Packet{}
nb := make([]byte, 12)
li.ListenOutBatch(func(addrs []netip.AddrPort, payloads [][]byte, count int) {
f.readOutsidePacketsBatch(addrs, payloads, count, outs[:count], nb, i, h, fwPacket, lhh, ctCache.Get(f.l))
nb := make([]byte, 12, 12)
li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get(f.l))
})
}
func (f *Interface) listenIn(reader overlay.BatchReadWriter, i int) {
func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
runtime.LockOSThread()
batchSize := reader.BatchSize()
// Allocate buffers for batch reading
bufs := make([][]byte, batchSize)
for idx := range bufs {
bufs[idx] = make([]byte, mtu)
}
sizes := make([]int, batchSize)
// Allocate output buffers for batch processing (one per packet)
// Each has virtio header headroom to avoid copies on write
outs := make([][]byte, batchSize)
for idx := range outs {
outBuf := make([]byte, virtioNetHdrLen+mtu)
outs[idx] = outBuf[virtioNetHdrLen:] // Slice starting after headroom
}
// Pre-allocate batch accumulation buffers for sending
batchPackets := make([][]byte, 0, batchSize)
batchAddrs := make([]netip.AddrPort, 0, batchSize)
// Pre-allocate nonce buffer (reused for all encryptions)
nb := make([]byte, 12)
packet := make([]byte, mtu)
out := make([]byte, mtu)
fwPacket := &firewall.Packet{}
nb := make([]byte, 12, 12)
conntrackCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
for {
n, err := reader.BatchRead(bufs, sizes)
n, err := reader.Read(packet)
if err != nil {
if errors.Is(err, os.ErrClosed) && f.closed.Load() {
return
}
f.l.WithError(err).Error("Error while batch reading outbound packets")
f.l.WithError(err).Error("Error while reading outbound packet")
// This only seems to happen when something fatal happens to the fd, so exit.
os.Exit(2)
}
// Process all packets in the batch at once
f.consumeInsidePackets(bufs, sizes, n, outs, nb, i, conntrackCache.Get(f.l), &batchPackets, &batchAddrs)
f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i, conntrackCache.Get(f.l))
}
}
+3 -10
View File
@@ -713,21 +713,14 @@ func (lh *LightHouse) unlockedShouldAddV6(vpnAddr netip.Addr, to *V6AddrPort) bo
func (lh *LightHouse) IsLighthouseAddr(vpnAddr netip.Addr) bool {
l := lh.GetLighthouses()
for i := range l {
if l[i] == vpnAddr {
return true
}
}
return false
return slices.Contains(l, vpnAddr)
}
func (lh *LightHouse) IsAnyLighthouseAddr(vpnAddrs []netip.Addr) bool {
l := lh.GetLighthouses()
for i := range vpnAddrs {
for j := range l {
if l[j] == vpnAddrs[i] {
return true
}
if slices.Contains(l, vpnAddrs[i]) {
return true
}
}
return false
+7 -10
View File
@@ -105,11 +105,7 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
// deprecated and undocumented
tunQueues := c.GetInt("tun.routines", 1)
udpQueues := c.GetInt("listen.routines", 1)
if tunQueues > udpQueues {
routines = tunQueues
} else {
routines = udpQueues
}
routines = max(tunQueues, udpQueues)
if routines != 1 {
l.WithField("routines", routines).Warn("Setting tun.routines and listen.routines is deprecated. Use `routines` instead")
}
@@ -171,7 +167,7 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
for i := 0; i < routines; i++ {
l.Infof("listening on %v", netip.AddrPortFrom(listenHost, uint16(port)))
udpServer, err := udp.NewListener(l, listenHost, port, routines > 1, c.GetInt("listen.batch", 128))
udpServer, err := udp.NewListener(l, listenHost, port, routines > 1, c.GetInt("listen.batch", 64))
if err != nil {
return nil, util.NewContextualError("Failed to open udp listener", m{"queue": i}, err)
}
@@ -208,10 +204,11 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
useRelays := c.GetBool("relay.use_relays", DefaultUseRelays) && !c.GetBool("relay.am_relay", false)
handshakeConfig := HandshakeConfig{
tryInterval: c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
retries: int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
triggerBuffer: c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
useRelays: useRelays,
tryInterval: c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
retries: int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
triggerBuffer: c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
useRelays: useRelays,
maxHandshakeRate: c.GetInt("handshakes.max_rate", DefaultMaxHandshakeRate),
messageMetrics: messageMetrics,
}
-1
View File
@@ -1,5 +1,4 @@
//go:build !boringcrypto
// +build !boringcrypto
package nebula
+53 -133
View File
@@ -102,7 +102,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
relay: relay,
IsRelayed: true,
}
f.readOutsidePackets(via, out[:virtioNetHdrLen], signedPayload, h, fwPacket, lhf, nb, q, localCache)
f.readOutsidePackets(via, out[:0], signedPayload, h, fwPacket, lhf, nb, q, localCache)
return
case ForwardingType:
// Find the target HostInfo relay object
@@ -145,7 +145,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
}
//TODO: assert via is not relayed
lhf.HandleRequest(via.UdpAddr, hostinfo.vpnAddrs, d[virtioNetHdrLen:], f)
lhf.HandleRequest(via.UdpAddr, hostinfo.vpnAddrs, d, f)
// Fallthrough to the bottom to record incoming traffic
@@ -167,7 +167,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
// This testRequest might be from TryPromoteBest, so we should roam
// to the new IP address before responding
f.handleHostRoaming(hostinfo, via)
f.send(header.Test, header.TestReply, ci, hostinfo, d[virtioNetHdrLen:], nb, out)
f.send(header.Test, header.TestReply, ci, hostinfo, d, nb, out)
}
// Fallthrough to the bottom to record incoming traffic
@@ -190,6 +190,13 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
if !f.handleEncrypted(ci, via, h) {
return
}
_, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb)
if err != nil {
hostinfo.logger(f.l).WithError(err).WithField("from", via).
WithField("packet", packet).
Error("Failed to decrypt CloseTunnel packet")
return
}
hostinfo.logger(f.l).WithField("from", via).
Info("Close tunnel received, tearing down.")
@@ -210,7 +217,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
return
}
f.relayManager.HandleControlMsg(hostinfo, d[virtioNetHdrLen:], f)
f.relayManager.HandleControlMsg(hostinfo, d, f)
default:
f.messageMetrics.Rx(h.Type, h.Subtype, 1)
@@ -327,13 +334,29 @@ func parseV6(data []byte, incoming bool, fp *firewall.Packet) error {
proto := layers.IPProtocol(data[protoAt])
switch proto {
case layers.IPProtocolICMPv6, layers.IPProtocolESP, layers.IPProtocolNoNextHeader:
case layers.IPProtocolESP, layers.IPProtocolNoNextHeader:
fp.Protocol = uint8(proto)
fp.RemotePort = 0
fp.LocalPort = 0
fp.Fragment = false
return nil
case layers.IPProtocolICMPv6:
if dataLen < offset+6 {
return ErrIPv6PacketTooShort
}
fp.Protocol = uint8(proto)
fp.LocalPort = 0 //incoming vs outgoing doesn't matter for icmpv6
icmptype := data[offset+1]
switch icmptype {
case layers.ICMPv6TypeEchoRequest, layers.ICMPv6TypeEchoReply:
fp.RemotePort = binary.BigEndian.Uint16(data[offset+4 : offset+6]) //identifier
default:
fp.RemotePort = 0
}
fp.Fragment = false
return nil
case layers.IPProtocolTCP, layers.IPProtocolUDP:
if dataLen < offset+4 {
return ErrIPv6PacketTooShort
@@ -423,34 +446,38 @@ func parseV4(data []byte, incoming bool, fp *firewall.Packet) error {
// Accounting for a variable header length, do we have enough data for our src/dst tuples?
minLen := ihl
if !fp.Fragment && fp.Protocol != firewall.ProtoICMP {
minLen += minFwPacketLen
if !fp.Fragment {
if fp.Protocol == firewall.ProtoICMP {
minLen += minFwPacketLen + 2
} else {
minLen += minFwPacketLen
}
}
if len(data) < minLen {
return ErrIPv4InvalidHeaderLength
}
// Firewall packets are locally oriented
if incoming {
if incoming { // Firewall packets are locally oriented
fp.RemoteAddr, _ = netip.AddrFromSlice(data[12:16])
fp.LocalAddr, _ = netip.AddrFromSlice(data[16:20])
if fp.Fragment || fp.Protocol == firewall.ProtoICMP {
fp.RemotePort = 0
fp.LocalPort = 0
} else {
fp.RemotePort = binary.BigEndian.Uint16(data[ihl : ihl+2])
fp.LocalPort = binary.BigEndian.Uint16(data[ihl+2 : ihl+4])
}
} else {
fp.LocalAddr, _ = netip.AddrFromSlice(data[12:16])
fp.RemoteAddr, _ = netip.AddrFromSlice(data[16:20])
if fp.Fragment || fp.Protocol == firewall.ProtoICMP {
fp.RemotePort = 0
fp.LocalPort = 0
} else {
fp.LocalPort = binary.BigEndian.Uint16(data[ihl : ihl+2])
fp.RemotePort = binary.BigEndian.Uint16(data[ihl+2 : ihl+4])
}
}
if fp.Fragment {
fp.RemotePort = 0
fp.LocalPort = 0
} else if fp.Protocol == firewall.ProtoICMP { //note that orientation doesn't matter on ICMP
fp.RemotePort = binary.BigEndian.Uint16(data[ihl+4 : ihl+6]) //identifier
fp.LocalPort = 0 //code would be uint16(data[ihl+1])
} else if incoming {
fp.RemotePort = binary.BigEndian.Uint16(data[ihl : ihl+2]) //src port
fp.LocalPort = binary.BigEndian.Uint16(data[ihl+2 : ihl+4]) //dst port
} else {
fp.LocalPort = binary.BigEndian.Uint16(data[ihl : ihl+2]) //src port
fp.RemotePort = binary.BigEndian.Uint16(data[ihl+2 : ihl+4]) //dst port
}
return nil
@@ -481,11 +508,9 @@ func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out
return false
}
packetData := out[virtioNetHdrLen:]
err = newPacket(packetData, true, fwPacket)
err = newPacket(out, true, fwPacket)
if err != nil {
hostinfo.logger(f.l).WithError(err).WithField("packet", packetData).
hostinfo.logger(f.l).WithError(err).WithField("packet", out).
Warnf("Error while validating inbound packet")
return false
}
@@ -500,7 +525,7 @@ func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out
if dropReason != nil {
// NOTE: We give `packet` as the `out` here since we already decrypted from it and we don't need it anymore
// This gives us a buffer to build the reject packet in
f.rejectOutside(packetData, hostinfo.ConnectionState, hostinfo, nb, packet, q)
f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, nb, packet, q)
if f.l.Level >= logrus.DebugLevel {
hostinfo.logger(f.l).WithField("fwPacket", fwPacket).
WithField("reason", dropReason).
@@ -564,108 +589,3 @@ func (f *Interface) handleRecvError(addr netip.AddrPort, h *header.H) {
// We also delete it from pending hostmap to allow for fast reconnect.
f.handshakeManager.DeleteHostInfo(hostinfo)
}
// readOutsidePacketsBatch processes multiple packets received from UDP in a batch
// and writes all successfully decrypted packets to TUN in a single operation
func (f *Interface) readOutsidePacketsBatch(addrs []netip.AddrPort, payloads [][]byte, count int, outs [][]byte, nb []byte, q int, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, localCache firewall.ConntrackCache) {
// Pre-allocate slice for accumulating successful decryptions
tunPackets := make([][]byte, 0, count)
for i := 0; i < count; i++ {
payload := payloads[i]
addr := addrs[i]
out := outs[i]
// Parse header
err := h.Parse(payload)
if err != nil {
if len(payload) > 1 {
f.l.WithField("packet", payload).Infof("Error while parsing inbound packet from %s: %s", addr, err)
}
continue
}
if addr.IsValid() {
if f.myVpnNetworksTable.Contains(addr.Addr()) {
if f.l.Level >= logrus.DebugLevel {
f.l.WithField("udpAddr", addr).Debug("Refusing to process double encrypted packet")
}
continue
}
}
var hostinfo *HostInfo
if h.Type == header.Message && h.Subtype == header.MessageRelay {
hostinfo = f.hostMap.QueryRelayIndex(h.RemoteIndex)
} else {
hostinfo = f.hostMap.QueryIndex(h.RemoteIndex)
}
var ci *ConnectionState
if hostinfo != nil {
ci = hostinfo.ConnectionState
}
switch h.Type {
case header.Message:
if !f.handleEncrypted(ci, ViaSender{UdpAddr: addr}, h) {
continue
}
switch h.Subtype {
case header.MessageNone:
// Decrypt packet
out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, payload[:header.Len], payload[header.Len:], h.MessageCounter, nb)
if err != nil {
hostinfo.logger(f.l).WithError(err).Error("Failed to decrypt packet")
continue
}
packetData := out[virtioNetHdrLen:]
err = newPacket(packetData, true, fwPacket)
if err != nil {
hostinfo.logger(f.l).WithError(err).WithField("packet", packetData).Warnf("Error while validating inbound packet")
continue
}
if !hostinfo.ConnectionState.window.Update(f.l, h.MessageCounter) {
hostinfo.logger(f.l).WithField("fwPacket", fwPacket).Debugln("dropping out of window packet")
continue
}
dropReason := f.firewall.Drop(*fwPacket, true, hostinfo, f.pki.GetCAPool(), localCache)
if dropReason != nil {
f.rejectOutside(packetData, hostinfo.ConnectionState, hostinfo, nb, payload, q)
if f.l.Level >= logrus.DebugLevel {
hostinfo.logger(f.l).WithField("fwPacket", fwPacket).WithField("reason", dropReason).Debugln("dropping inbound packet")
}
continue
}
f.connectionManager.In(hostinfo)
// Add to batch for TUN write
tunPackets = append(tunPackets, out)
case header.MessageRelay:
// Skip relay packets in batch mode for now (less common path)
f.readOutsidePackets(ViaSender{UdpAddr: addr}, out[:virtioNetHdrLen], payload, h, fwPacket, lhf, nb, q, localCache)
default:
hostinfo.logger(f.l).Debugf("unexpected message subtype %d", h.Subtype)
}
default:
// Handle non-Message types using single-packet path
f.readOutsidePackets(ViaSender{UdpAddr: addr}, out[:virtioNetHdrLen], payload, h, fwPacket, lhf, nb, q, localCache)
}
}
if len(tunPackets) > 0 {
n, err := f.readers[q].WriteBatch(tunPackets, virtioNetHdrLen)
if err != nil {
f.l.WithError(err).WithField("sent", n).WithField("total", len(tunPackets)).Error("Failed to batch write to tun")
}
}
}
+17 -10
View File
@@ -155,6 +155,7 @@ func Test_newPacket_v6(t *testing.T) {
// next layer, missing length byte
err = newPacket(buffer.Bytes()[:49], true, p)
require.ErrorIs(t, err, ErrIPv6CouldNotFindPayload)
err = nil
// A good ICMP packet
ip = layers.IPv6{
@@ -165,20 +166,26 @@ func Test_newPacket_v6(t *testing.T) {
DstIP: net.IPv6linklocalallnodes,
}
icmp := layers.ICMPv6{}
buffer.Clear()
err = gopacket.SerializeLayers(buffer, opt, &ip, &icmp)
if err != nil {
panic(err)
icmp := layers.ICMPv6{
TypeCode: layers.ICMPv6TypeEchoRequest,
Checksum: 0x1234,
}
err = newPacket(buffer.Bytes(), true, p)
require.NoError(t, err)
buffer.Clear()
require.NoError(t, gopacket.SerializeLayers(buffer, opt, &ip, &icmp))
require.Error(t, newPacket(buffer.Bytes(), true, p))
buffer.Clear()
echo := layers.ICMPv6Echo{
Identifier: 0xabcd,
SeqNumber: 1234,
}
require.NoError(t, gopacket.SerializeLayers(buffer, opt, &ip, &icmp, &echo))
require.NoError(t, newPacket(buffer.Bytes(), true, p))
assert.Equal(t, uint8(layers.IPProtocolICMPv6), p.Protocol)
assert.Equal(t, netip.MustParseAddr("ff02::2"), p.RemoteAddr)
assert.Equal(t, netip.MustParseAddr("ff02::1"), p.LocalAddr)
assert.Equal(t, uint16(0), p.RemotePort)
assert.Equal(t, uint16(0xabcd), p.RemotePort)
assert.Equal(t, uint16(0), p.LocalPort)
assert.False(t, p.Fragment)
@@ -574,7 +581,7 @@ func BenchmarkParseV6(b *testing.B) {
}
evilBytes := buffer.Bytes()
for i := 0; i < 200; i++ {
for range 200 {
evilBytes = append(evilBytes, hopHeader...)
}
evilBytes = append(evilBytes, lastHopHeader...)
+2 -16
View File
@@ -7,26 +7,12 @@ import (
"github.com/slackhq/nebula/routing"
)
// BatchReadWriter extends io.ReadWriteCloser with batch I/O operations
type BatchReadWriter interface {
io.ReadWriteCloser
// BatchRead reads multiple packets at once
BatchRead(bufs [][]byte, sizes []int) (int, error)
// WriteBatch writes multiple packets at once
WriteBatch(bufs [][]byte, offset int) (int, error)
// BatchSize returns the optimal batch size for this device
BatchSize() int
}
type Device interface {
BatchReadWriter
io.ReadWriteCloser
Activate() error
Networks() []netip.Prefix
Name() string
RoutesFor(netip.Addr) routing.Gateways
SupportsMultiqueue() bool
NewMultiQueueReader() (BatchReadWriter, error)
NewMultiQueueReader() (io.ReadWriteCloser, error)
}
-1
View File
@@ -11,7 +11,6 @@ import (
)
const DefaultMTU = 1300
const VirtioNetHdrLen = 10 // Size of virtio_net_hdr structure
type NameError struct {
Name string
+1 -24
View File
@@ -99,29 +99,6 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for android")
}
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
func (t *tun) BatchSize() int {
return 1
}
+1 -27
View File
@@ -553,32 +553,6 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for darwin")
}
// BatchRead reads a single packet (batch size 1 for non-Linux platforms)
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// WriteBatch writes packets individually (no batching for non-Linux platforms)
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
// BatchSize returns 1 for non-Linux platforms (no batching)
func (t *tun) BatchSize() int {
return 1
}
+1 -27
View File
@@ -109,36 +109,10 @@ func (t *disabledTun) SupportsMultiqueue() bool {
return true
}
func (t *disabledTun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *disabledTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return t, nil
}
// BatchRead reads a single packet (batch size 1 for disabled tun)
func (t *disabledTun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// WriteBatch writes packets individually (no batching for disabled tun)
func (t *disabledTun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
// BatchSize returns 1 for disabled tun (no batching)
func (t *disabledTun) BatchSize() int {
return 1
}
func (t *disabledTun) Close() error {
if t.read != nil {
close(t.read)
+2 -27
View File
@@ -7,6 +7,7 @@ import (
"bytes"
"errors"
"fmt"
"io"
"io/fs"
"net/netip"
"sync/atomic"
@@ -453,36 +454,10 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for freebsd")
}
// BatchRead reads a single packet (batch size 1 for FreeBSD)
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// WriteBatch writes packets individually (no batching for FreeBSD)
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
// BatchSize returns 1 for FreeBSD (no batching)
func (t *tun) BatchSize() int {
return 1
}
func (t *tun) addRoutes(logErrors bool) error {
routes := *t.Routes.Load()
for _, r := range routes {
+1 -24
View File
@@ -155,29 +155,6 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for ios")
}
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
func (t *tun) BatchSize() int {
return 1
}
+32 -227
View File
@@ -9,6 +9,7 @@ import (
"net"
"net/netip"
"os"
"strings"
"sync"
"sync/atomic"
"time"
@@ -21,12 +22,10 @@ import (
"github.com/slackhq/nebula/util"
"github.com/vishvananda/netlink"
"golang.org/x/sys/unix"
wgtun "golang.zx2c4.com/wireguard/tun"
)
type tun struct {
io.ReadWriteCloser
wgDevice wgtun.Device
fd int
Device string
vpnNetworks []netip.Prefix
@@ -72,154 +71,63 @@ type ifreqQLEN struct {
pad [8]byte
}
// wgDeviceWrapper wraps a wireguard Device to implement io.ReadWriteCloser
// This allows multiqueue readers to use the same wireguard Device batching as the main device
type wgDeviceWrapper struct {
dev wgtun.Device
buf []byte // Reusable buffer for single packet reads
}
func (w *wgDeviceWrapper) Read(b []byte) (int, error) {
// Use wireguard Device's batch API for single packet
bufs := [][]byte{b}
sizes := make([]int, 1)
n, err := w.dev.Read(bufs, sizes, 0)
if err != nil {
return 0, err
}
if n == 0 {
return 0, io.EOF
}
return sizes[0], nil
}
func (w *wgDeviceWrapper) Write(b []byte) (int, error) {
// Buffer b should have virtio header space (10 bytes) at the beginning
// The decrypted packet data starts at offset 10
// Pass the full buffer to WireGuard with offset=virtioNetHdrLen
bufs := [][]byte{b}
n, err := w.dev.Write(bufs, VirtioNetHdrLen)
if err != nil {
return 0, err
}
if n == 0 {
return 0, io.ErrShortWrite
}
return len(b), nil
}
func (w *wgDeviceWrapper) WriteBatch(bufs [][]byte, offset int) (int, error) {
// Pass all buffers to WireGuard's batch write
return w.dev.Write(bufs, offset)
}
func (w *wgDeviceWrapper) Close() error {
return w.dev.Close()
}
// BatchRead implements batching for multiqueue readers
func (w *wgDeviceWrapper) BatchRead(bufs [][]byte, sizes []int) (int, error) {
// The zero here is offset.
return w.dev.Read(bufs, sizes, 0)
}
// BatchSize returns the optimal batch size
func (w *wgDeviceWrapper) BatchSize() int {
return w.dev.BatchSize()
}
func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
wgDev, name, err := wgtun.CreateUnmonitoredTUNFromFD(deviceFd)
if err != nil {
return nil, fmt.Errorf("failed to create TUN from FD: %w", err)
}
file := os.NewFile(uintptr(deviceFd), "/dev/net/tun")
file := wgDev.File()
t, err := newTunGeneric(c, l, file, vpnNetworks)
if err != nil {
_ = wgDev.Close()
return nil, err
}
t.wgDevice = wgDev
t.Device = name
t.Device = "tun0"
return t, nil
}
func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) {
// Check if /dev/net/tun exists, create if needed (for docker containers)
if _, err := os.Stat("/dev/net/tun"); os.IsNotExist(err) {
if err := os.MkdirAll("/dev/net", 0755); err != nil {
return nil, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
}
if err := unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200))); err != nil {
return nil, fmt.Errorf("failed to create /dev/net/tun: %w", err)
}
}
devName := c.GetString("tun.dev", "")
mtu := c.GetInt("tun.mtu", DefaultMTU)
// Create TUN device manually to support multiqueue
fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
if err != nil {
return nil, err
// If /dev/net/tun doesn't exist, try to create it (will happen in docker)
if os.IsNotExist(err) {
err = os.MkdirAll("/dev/net", 0755)
if err != nil {
return nil, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
}
err = unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200)))
if err != nil {
return nil, fmt.Errorf("failed to create /dev/net/tun: %w", err)
}
fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
if err != nil {
return nil, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
}
} else {
return nil, err
}
}
var req ifReq
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_VNET_HDR)
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI)
if multiqueue {
req.Flags |= unix.IFF_MULTI_QUEUE
}
copy(req.Name[:], devName)
nameStr := c.GetString("tun.dev", "")
copy(req.Name[:], nameStr)
if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
unix.Close(fd)
return nil, err
}
// Set nonblocking
if err = unix.SetNonblock(fd, true); err != nil {
unix.Close(fd)
return nil, err
}
// Enable TCP and UDP offload (TSO/GRO) for performance
// This allows the kernel to handle segmentation/coalescing
const (
tunTCPOffloads = unix.TUN_F_CSUM | unix.TUN_F_TSO4 | unix.TUN_F_TSO6
tunUDPOffloads = unix.TUN_F_USO4 | unix.TUN_F_USO6
)
offloads := tunTCPOffloads | tunUDPOffloads
if err = unix.IoctlSetInt(fd, unix.TUNSETOFFLOAD, offloads); err != nil {
// Log warning but don't fail - offload is optional
l.WithError(err).Warn("Failed to enable TUN offload (TSO/GRO), performance may be reduced")
return nil, &NameError{
Name: nameStr,
Underlying: err,
}
}
name := strings.Trim(string(req.Name[:]), "\x00")
file := os.NewFile(uintptr(fd), "/dev/net/tun")
// Create wireguard device from file descriptor
wgDev, err := wgtun.CreateTUNFromFile(file, mtu)
if err != nil {
file.Close()
return nil, fmt.Errorf("failed to create TUN from file: %w", err)
}
name, err := wgDev.Name()
if err != nil {
_ = wgDev.Close()
return nil, fmt.Errorf("failed to get TUN device name: %w", err)
}
// file is now owned by wgDev, get a new reference
file = wgDev.File()
t, err := newTunGeneric(c, l, file, vpnNetworks)
if err != nil {
_ = wgDev.Close()
return nil, err
}
t.wgDevice = wgDev
t.Device = name
return t, nil
@@ -330,44 +238,22 @@ func (t *tun) SupportsMultiqueue() bool {
return true
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
if err != nil {
return nil, err
}
var req ifReq
// MUST match the flags used in newTun - includes IFF_VNET_HDR
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_VNET_HDR | unix.IFF_MULTI_QUEUE)
req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
copy(req.Name[:], t.Device)
if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
unix.Close(fd)
return nil, err
}
// Set nonblocking mode - CRITICAL for proper netpoller integration
if err = unix.SetNonblock(fd, true); err != nil {
unix.Close(fd)
return nil, err
}
// Get MTU from main device
mtu := t.MaxMTU
if mtu == 0 {
mtu = DefaultMTU
}
file := os.NewFile(uintptr(fd), "/dev/net/tun")
// Create wireguard Device from the file descriptor (just like the main device)
wgDev, err := wgtun.CreateTUNFromFile(file, mtu)
if err != nil {
file.Close()
return nil, fmt.Errorf("failed to create multiqueue TUN device: %w", err)
}
// Return a wrapper that uses the wireguard Device for all I/O
return &wgDeviceWrapper{dev: wgDev}, nil
return file, nil
}
func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
@@ -375,68 +261,7 @@ func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
return r
}
func (t *tun) Read(b []byte) (int, error) {
if t.wgDevice != nil {
// Use wireguard device which handles virtio headers internally
bufs := [][]byte{b}
sizes := make([]int, 1)
n, err := t.wgDevice.Read(bufs, sizes, 0)
if err != nil {
return 0, err
}
if n == 0 {
return 0, io.EOF
}
return sizes[0], nil
}
// Fallback: direct read from file (shouldn't happen in normal operation)
return t.ReadWriteCloser.Read(b)
}
// BatchRead reads multiple packets at once for improved performance
// bufs: slice of buffers to read into
// sizes: slice that will be filled with packet sizes
// Returns number of packets read
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
if t.wgDevice != nil {
return t.wgDevice.Read(bufs, sizes, 0)
}
// Fallback: single packet read
n, err := t.ReadWriteCloser.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// BatchSize returns the optimal number of packets to read/write in a batch
func (t *tun) BatchSize() int {
if t.wgDevice != nil {
return t.wgDevice.BatchSize()
}
return 1
}
func (t *tun) Write(b []byte) (int, error) {
if t.wgDevice != nil {
// Buffer b should have virtio header space (10 bytes) at the beginning
// The decrypted packet data starts at offset 10
// Pass the full buffer to WireGuard with offset=virtioNetHdrLen
bufs := [][]byte{b}
n, err := t.wgDevice.Write(bufs, VirtioNetHdrLen)
if err != nil {
return 0, err
}
if n == 0 {
return 0, io.ErrShortWrite
}
return len(b), nil
}
// Fallback: direct write (shouldn't happen in normal operation)
var nn int
maximum := len(b)
@@ -459,22 +284,6 @@ func (t *tun) Write(b []byte) (int, error) {
}
}
// WriteBatch writes multiple packets to the TUN device in a single syscall
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
if t.wgDevice != nil {
return t.wgDevice.Write(bufs, offset)
}
// Fallback: write individually (shouldn't happen in normal operation)
for i, buf := range bufs {
_, err := t.Write(buf)
if err != nil {
return i, err
}
}
return len(bufs), nil
}
func (t *tun) deviceBytes() (o [16]byte) {
for i, c := range t.Device {
o[i] = byte(c)
@@ -902,10 +711,6 @@ func (t *tun) Close() error {
close(t.routeChan)
}
if t.wgDevice != nil {
_ = t.wgDevice.Close()
}
if t.ReadWriteCloser != nil {
_ = t.ReadWriteCloser.Close()
}
+2 -24
View File
@@ -6,6 +6,7 @@ package overlay
import (
"errors"
"fmt"
"io"
"net/netip"
"os"
"regexp"
@@ -393,33 +394,10 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for netbsd")
}
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
func (t *tun) BatchSize() int {
return 1
}
func (t *tun) addRoutes(logErrors bool) error {
routes := *t.Routes.Load()
+2 -24
View File
@@ -6,6 +6,7 @@ package overlay
import (
"errors"
"fmt"
"io"
"net/netip"
"os"
"regexp"
@@ -313,33 +314,10 @@ func (t *tun) SupportsMultiqueue() bool {
return false
}
func (t *tun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for openbsd")
}
func (t *tun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
func (t *tun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
func (t *tun) BatchSize() int {
return 1
}
func (t *tun) addRoutes(logErrors bool) error {
routes := *t.Routes.Load()
+3 -35
View File
@@ -109,12 +109,8 @@ func (t *TestTun) Write(b []byte) (n int, err error) {
return 0, io.ErrClosedPipe
}
// Skip virtio header (consistent with production Linux tun)
// The buffer b has VirtioNetHdrLen bytes of header followed by the actual packet
data := b[VirtioNetHdrLen:]
packet := make([]byte, len(data))
copy(packet, data)
packet := make([]byte, len(b), len(b))
copy(packet, b)
t.TxPackets <- packet
return len(b), nil
}
@@ -140,34 +136,6 @@ func (t *TestTun) SupportsMultiqueue() bool {
return false
}
func (t *TestTun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *TestTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented")
}
func (t *TestTun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
func (t *TestTun) WriteBatch(bufs [][]byte, offset int) (int, error) {
if t.closed.Load() {
return 0, io.ErrClosedPipe
}
for _, buf := range bufs {
// Strip the header at offset and send directly to channel
data := buf[offset:]
packet := make([]byte, len(data))
copy(packet, data)
t.TxPackets <- packet
}
return len(bufs), nil
}
func (t *TestTun) BatchSize() int {
return 1
}
+2 -27
View File
@@ -6,6 +6,7 @@ package overlay
import (
"crypto"
"fmt"
"io"
"net/netip"
"os"
"path/filepath"
@@ -240,36 +241,10 @@ func (t *winTun) SupportsMultiqueue() bool {
return false
}
func (t *winTun) NewMultiQueueReader() (BatchReadWriter, error) {
func (t *winTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, fmt.Errorf("TODO: multiqueue not implemented for windows")
}
// BatchRead reads a single packet (batch size 1 for Windows)
func (t *winTun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := t.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// WriteBatch writes packets individually (no batching for Windows)
func (t *winTun) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := t.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
// BatchSize returns 1 for Windows (no batching)
func (t *winTun) BatchSize() int {
return 1
}
func (t *winTun) Close() error {
// It seems that the Windows networking stack doesn't like it when we destroy interfaces that have active routes,
// so to be certain, just remove everything before destroying.
+1 -27
View File
@@ -50,36 +50,10 @@ func (d *UserDevice) SupportsMultiqueue() bool {
return true
}
func (d *UserDevice) NewMultiQueueReader() (BatchReadWriter, error) {
func (d *UserDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return d, nil
}
// BatchRead reads a single packet (batch size 1 for UserDevice)
func (d *UserDevice) BatchRead(bufs [][]byte, sizes []int) (int, error) {
n, err := d.Read(bufs[0])
if err != nil {
return 0, err
}
sizes[0] = n
return 1, nil
}
// WriteBatch writes packets individually (no batching for UserDevice)
func (d *UserDevice) WriteBatch(bufs [][]byte, offset int) (int, error) {
for i, buf := range bufs {
_, err := d.Write(buf[offset:])
if err != nil {
return i, err
}
}
return len(bufs), nil
}
// BatchSize returns 1 for UserDevice (no batching)
func (d *UserDevice) BatchSize() int {
return 1
}
func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
return d.inboundReader, d.outboundWriter
}
+1 -1
View File
@@ -55,7 +55,7 @@ func (rm *relayManager) setAmRelay(v bool) {
func AddRelay(l *logrus.Logger, relayHostInfo *HostInfo, hm *HostMap, vpnIp netip.Addr, remoteIdx *uint32, relayType int, state int) (uint32, error) {
hm.Lock()
defer hm.Unlock()
for i := 0; i < 32; i++ {
for range 32 {
index, err := generateIndex(l)
if err != nil {
return 0, err
+1 -6
View File
@@ -404,12 +404,7 @@ func (r *RemoteList) Rebuild(preferredRanges []netip.Prefix) {
// unlockedIsBad assumes you have the write lock and checks if the remote matches any entry in the blocked address list
func (r *RemoteList) unlockedIsBad(remote netip.AddrPort) bool {
for _, v := range r.badRemotes {
if v == remote {
return true
}
}
return false
return slices.Contains(r.badRemotes, remote)
}
// unlockedSetLearnedV4 assumes you have the write lock and sets the current learned address for this owner and marks the
+5 -3
View File
@@ -6,6 +6,7 @@ import (
"errors"
"flag"
"fmt"
"maps"
"net"
"net/netip"
"os"
@@ -631,6 +632,9 @@ func sshCreateTunnel(ifce *Interface, fs any, a []string, w sshd.StringWriter) e
}
hostInfo = ifce.handshakeManager.StartHandshake(vpnAddr, nil)
if hostInfo == nil {
return w.WriteLine("Handshake rate limit reached")
}
if addr.IsValid() {
hostInfo.SetRemote(addr)
}
@@ -831,9 +835,7 @@ func sshPrintRelays(ifce *Interface, fs any, a []string, w sshd.StringWriter) er
relays := map[uint32]*HostInfo{}
ifce.hostMap.Lock()
for k, v := range ifce.hostMap.Relays {
relays[k] = v
}
maps.Copy(relays, ifce.hostMap.Relays)
ifce.hostMap.Unlock()
type RelayFor struct {
-1
View File
@@ -6,7 +6,6 @@ import (
"log"
"net"
"net/http"
_ "net/http/pprof"
"runtime"
"strconv"
"time"
+2 -18
View File
@@ -1,11 +1,10 @@
package device
package test
import (
"errors"
"io"
"net/netip"
"github.com/slackhq/nebula/overlay"
"github.com/slackhq/nebula/routing"
)
@@ -39,25 +38,10 @@ func (NoopTun) SupportsMultiqueue() bool {
return false
}
func (NoopTun) NewMultiQueueReader() (overlay.BatchReadWriter, error) {
func (NoopTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
return nil, errors.New("unsupported")
}
func (NoopTun) Close() error {
return nil
}
// BatchRead implements BatchReadWriter interface
func (NoopTun) BatchRead(bufs [][]byte, sizes []int) (int, error) {
return 0, io.EOF
}
// WriteBatch implements BatchReadWriter interface
func (NoopTun) WriteBatch(bufs [][]byte, offset int) (int, error) {
return len(bufs), nil
}
// BatchSize implements BatchReadWriter interface
func (NoopTun) BatchSize() int {
return 1
}
+2 -2
View File
@@ -134,7 +134,7 @@ func TestTimerWheel_Purge(t *testing.T) {
assert.True(t, tw.lastTick.After(lastTick))
// Make sure we get all 4 packets back
for i := 0; i < 4; i++ {
for i := range 4 {
p, has := tw.Purge()
assert.True(t, has)
assert.Equal(t, fps[i], p)
@@ -149,7 +149,7 @@ func TestTimerWheel_Purge(t *testing.T) {
// Make sure we cached the free'd items
assert.Equal(t, 4, tw.itemsCached)
ci := tw.itemCache
for i := 0; i < 4; i++ {
for range 4 {
assert.NotNil(t, ci)
ci = ci.Next
}
+6 -22
View File
@@ -13,21 +13,13 @@ type EncReader func(
payload []byte,
)
type EncBatchReader func(
addrs []netip.AddrPort,
payloads [][]byte,
count int,
)
type Conn interface {
Rebind() error
LocalAddr() (netip.AddrPort, error)
ListenOutBatch(r EncBatchReader)
ListenOut(r EncReader)
WriteTo(b []byte, addr netip.AddrPort) error
WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error)
ReloadConfig(c *config.C)
SupportsMultipleReaders() bool
BatchSize() int
Close() error
}
@@ -39,25 +31,17 @@ func (NoopConn) Rebind() error {
func (NoopConn) LocalAddr() (netip.AddrPort, error) {
return netip.AddrPort{}, nil
}
func (NoopConn) ListenOut(_ EncReader) {}
func (NoopConn) ListenOut(_ EncReader) {
return
}
func (NoopConn) SupportsMultipleReaders() bool {
return false
}
func (NoopConn) ListenOutBatch(_ EncBatchReader) {}
func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error {
return nil
}
func (NoopConn) WriteMulti(_ [][]byte, _ []netip.AddrPort) (int, error) {
return 0, nil
}
func (NoopConn) ReloadConfig(_ *config.C) {}
func (NoopConn) BatchSize() int {
return 1
func (NoopConn) ReloadConfig(_ *config.C) {
return
}
func (NoopConn) Close() error {
return nil
-39
View File
@@ -140,17 +140,6 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
}
}
// WriteMulti sends multiple packets - fallback implementation without sendmmsg
func (u *StdConn) WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error) {
for i := range packets {
err := u.WriteTo(packets[i], addrs[i])
if err != nil {
return i, err
}
}
return len(packets), nil
}
func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
a := u.UDPConn.LocalAddr()
@@ -199,34 +188,6 @@ func (u *StdConn) SupportsMultipleReaders() bool {
return false
}
// ListenOutBatch - fallback to single-packet reads for Darwin
func (u *StdConn) ListenOutBatch(r EncBatchReader) {
buffer := make([]byte, MTU)
addrs := make([]netip.AddrPort, 1)
payloads := make([][]byte, 1)
for {
// Just read one packet at a time and call batch callback with count=1
n, rua, err := u.ReadFromUDPAddrPort(buffer)
if err != nil {
if errors.Is(err, net.ErrClosed) {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
}
u.l.WithError(err).Error("unexpected udp socket receive error")
}
addrs[0] = netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port())
payloads[0] = buffer[:n]
r(addrs, payloads, 1)
}
}
func (u *StdConn) BatchSize() int {
return 1
}
func (u *StdConn) Rebind() error {
var err error
if u.isV4 {
-35
View File
@@ -101,38 +101,3 @@ func (u *GenericConn) ListenOut(r EncReader) {
func (u *GenericConn) SupportsMultipleReaders() bool {
return false
}
// ListenOutBatch - fallback to single-packet reads for generic platforms
func (u *GenericConn) ListenOutBatch(r EncBatchReader) {
buffer := make([]byte, MTU)
addrs := make([]netip.AddrPort, 1)
payloads := make([][]byte, 1)
for {
// Just read one packet at a time and call batch callback with count=1
n, rua, err := u.ReadFromUDPAddrPort(buffer)
if err != nil {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
}
addrs[0] = netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port())
payloads[0] = buffer[:n]
r(addrs, payloads, 1)
}
}
// WriteMulti sends multiple packets - fallback implementation
func (u *GenericConn) WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error) {
for i := range packets {
err := u.WriteTo(packets[i], addrs[i])
if err != nil {
return i, err
}
}
return len(packets), nil
}
func (u *GenericConn) BatchSize() int {
return 1
}
+4 -174
View File
@@ -22,11 +22,6 @@ type StdConn struct {
isV4 bool
l *logrus.Logger
batch int
// Pre-allocated buffers for batch writes (sized for IPv6, works for both)
writeMsgs []rawMessage
writeIovecs []iovec
writeNames [][]byte
}
func maybeIPV4(ip net.IP) (net.IP, bool) {
@@ -74,26 +69,7 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
return nil, fmt.Errorf("unable to bind to socket: %s", err)
}
c := &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}
// Pre-allocate write message structures for batching (sized for IPv6, works for both)
c.writeMsgs = make([]rawMessage, batch)
c.writeIovecs = make([]iovec, batch)
c.writeNames = make([][]byte, batch)
for i := range c.writeMsgs {
// Allocate for IPv6 size (larger than IPv4, works for both)
c.writeNames[i] = make([]byte, unix.SizeofSockaddrInet6)
// Point to the iovec in the slice
c.writeMsgs[i].Hdr.Iov = &c.writeIovecs[i]
c.writeMsgs[i].Hdr.Iovlen = 1
c.writeMsgs[i].Hdr.Name = &c.writeNames[i][0]
// Namelen will be set appropriately in writeMulti4/writeMulti6
}
return c, err
return &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}, err
}
func (u *StdConn) SupportsMultipleReaders() bool {
@@ -146,7 +122,7 @@ func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
}
}
func (u *StdConn) ListenOutBatch(r EncBatchReader) {
func (u *StdConn) ListenOut(r EncReader) {
var ip netip.Addr
msgs, buffers, names := u.PrepareRawMessages(u.batch)
@@ -155,12 +131,6 @@ func (u *StdConn) ListenOutBatch(r EncBatchReader) {
read = u.ReadSingle
}
udpBatchHist := metrics.GetOrRegisterHistogram("batch.udp_read_size", nil, metrics.NewUniformSample(1024))
// Pre-allocate slices for batch callback
addrs := make([]netip.AddrPort, u.batch)
payloads := make([][]byte, u.batch)
for {
n, err := read(msgs)
if err != nil {
@@ -168,21 +138,15 @@ func (u *StdConn) ListenOutBatch(r EncBatchReader) {
return
}
udpBatchHist.Update(int64(n))
// Prepare batch data
for i := 0; i < n; i++ {
// Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic
if u.isV4 {
ip, _ = netip.AddrFromSlice(names[i][4:8])
} else {
ip, _ = netip.AddrFromSlice(names[i][8:24])
}
addrs[i] = netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4]))
payloads[i] = buffers[i][:msgs[i].Len]
r(netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4])), buffers[i][:msgs[i].Len])
}
// Call batch callback with all packets
r(addrs, payloads, n)
}
}
@@ -234,19 +198,6 @@ func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
return u.writeTo6(b, ip)
}
func (u *StdConn) WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error) {
if len(packets) != len(addrs) {
return 0, fmt.Errorf("packets and addrs length mismatch")
}
if len(packets) == 0 {
return 0, nil
}
if u.isV4 {
return u.writeMulti4(packets, addrs)
}
return u.writeMulti6(packets, addrs)
}
func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
var rsa unix.RawSockaddrInet6
rsa.Family = unix.AF_INET6
@@ -301,123 +252,6 @@ func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
}
}
func (u *StdConn) writeMulti4(packets [][]byte, addrs []netip.AddrPort) (int, error) {
sent := 0
for sent < len(packets) {
// Determine batch size based on remaining packets and buffer capacity
batchSize := len(packets) - sent
if batchSize > len(u.writeMsgs) {
batchSize = len(u.writeMsgs)
}
// Use pre-allocated buffers
msgs := u.writeMsgs[:batchSize]
iovecs := u.writeIovecs[:batchSize]
names := u.writeNames[:batchSize]
// Setup message structures for this batch
for i := 0; i < batchSize; i++ {
pktIdx := sent + i
if !addrs[pktIdx].Addr().Is4() {
return sent + i, ErrInvalidIPv6RemoteForSocket
}
// Setup the packet buffer
iovecs[i].Base = &packets[pktIdx][0]
iovecs[i].Len = uint(len(packets[pktIdx]))
// Setup the destination address
rsa := (*unix.RawSockaddrInet4)(unsafe.Pointer(&names[i][0]))
rsa.Family = unix.AF_INET
rsa.Addr = addrs[pktIdx].Addr().As4()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], addrs[pktIdx].Port())
// Set the appropriate address length for IPv4
msgs[i].Hdr.Namelen = unix.SizeofSockaddrInet4
}
// Send this batch
nsent, _, err := unix.Syscall6(
unix.SYS_SENDMMSG,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&msgs[0])),
uintptr(batchSize),
0,
0,
0,
)
if err != 0 {
return sent + int(nsent), &net.OpError{Op: "sendmmsg", Err: err}
}
sent += int(nsent)
if int(nsent) < batchSize {
// Couldn't send all packets in batch, return what we sent
return sent, nil
}
}
return sent, nil
}
func (u *StdConn) writeMulti6(packets [][]byte, addrs []netip.AddrPort) (int, error) {
sent := 0
for sent < len(packets) {
// Determine batch size based on remaining packets and buffer capacity
batchSize := len(packets) - sent
if batchSize > len(u.writeMsgs) {
batchSize = len(u.writeMsgs)
}
// Use pre-allocated buffers
msgs := u.writeMsgs[:batchSize]
iovecs := u.writeIovecs[:batchSize]
names := u.writeNames[:batchSize]
// Setup message structures for this batch
for i := 0; i < batchSize; i++ {
pktIdx := sent + i
// Setup the packet buffer
iovecs[i].Base = &packets[pktIdx][0]
iovecs[i].Len = uint(len(packets[pktIdx]))
// Setup the destination address
rsa := (*unix.RawSockaddrInet6)(unsafe.Pointer(&names[i][0]))
rsa.Family = unix.AF_INET6
rsa.Addr = addrs[pktIdx].Addr().As16()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], addrs[pktIdx].Port())
// Set the appropriate address length for IPv6
msgs[i].Hdr.Namelen = unix.SizeofSockaddrInet6
}
// Send this batch
nsent, _, err := unix.Syscall6(
unix.SYS_SENDMMSG,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&msgs[0])),
uintptr(batchSize),
0,
0,
0,
)
if err != 0 {
return sent + int(nsent), &net.OpError{Op: "sendmmsg", Err: err}
}
sent += int(nsent)
if int(nsent) < batchSize {
// Couldn't send all packets in batch, return what we sent
return sent, nil
}
}
return sent, nil
}
func (u *StdConn) ReloadConfig(c *config.C) {
b := c.GetInt("listen.read_buffer", 0)
if b > 0 {
@@ -475,10 +309,6 @@ func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
return nil
}
func (u *StdConn) BatchSize() int {
return u.batch
}
func (u *StdConn) Close() error {
return syscall.Close(u.sysFd)
}
+2 -2
View File
@@ -12,7 +12,7 @@ import (
type iovec struct {
Base *byte
Len uint
Len uint32
}
type msghdr struct {
@@ -40,7 +40,7 @@ func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
names[i] = make([]byte, unix.SizeofSockaddrInet6)
vs := []iovec{
{Base: &buffers[i][0], Len: uint(len(buffers[i]))},
{Base: &buffers[i][0], Len: uint32(len(buffers[i]))},
}
msgs[i].Hdr.Iov = &vs[0]
+2 -2
View File
@@ -12,7 +12,7 @@ import (
type iovec struct {
Base *byte
Len uint
Len uint64
}
type msghdr struct {
@@ -43,7 +43,7 @@ func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
names[i] = make([]byte, unix.SizeofSockaddrInet6)
vs := []iovec{
{Base: &buffers[i][0], Len: uint(len(buffers[i]))},
{Base: &buffers[i][0], Len: uint64(len(buffers[i]))},
}
msgs[i].Hdr.Iov = &vs[0]
-44
View File
@@ -338,50 +338,6 @@ func (u *RIOConn) Rebind() error {
func (u *RIOConn) ReloadConfig(*config.C) {}
// BatchSize returns 1 since RIO reads packets one at a time
func (u *RIOConn) BatchSize() int {
return 1
}
// ListenOutBatch - fallback to single-packet reads for RIO
func (u *RIOConn) ListenOutBatch(r EncBatchReader) {
buffer := make([]byte, MTU)
addrs := make([]netip.AddrPort, 1)
payloads := make([][]byte, 1)
var lastRecvErr time.Time
for {
n, rua, err := u.receive(buffer)
if err != nil {
if errors.Is(err, net.ErrClosed) {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
}
if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
lastRecvErr = time.Now()
u.l.WithError(err).Warn("unexpected udp socket receive error")
}
continue
}
addrs[0] = netip.AddrPortFrom(netip.AddrFrom16(rua.Addr).Unmap(), (rua.Port>>8)|((rua.Port&0xff)<<8))
payloads[0] = buffer[:n]
r(addrs, payloads, 1)
}
}
// WriteMulti sends multiple packets - fallback implementation
func (u *RIOConn) WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error) {
for i := range packets {
err := u.WriteTo(packets[i], addrs[i])
if err != nil {
return i, err
}
}
return len(packets), nil
}
func (u *RIOConn) Close() error {
if !u.isOpen.CompareAndSwap(true, false) {
return nil
-29
View File
@@ -116,31 +116,6 @@ func (u *TesterConn) ListenOut(r EncReader) {
}
}
func (u *TesterConn) ListenOutBatch(r EncBatchReader) {
addrs := make([]netip.AddrPort, 1)
payloads := make([][]byte, 1)
for {
p, ok := <-u.RxPackets
if !ok {
return
}
addrs[0] = p.From
payloads[0] = p.Data
r(addrs, payloads, 1)
}
}
func (u *TesterConn) WriteMulti(packets [][]byte, addrs []netip.AddrPort) (int, error) {
for i := range packets {
err := u.WriteTo(packets[i], addrs[i])
if err != nil {
return i, err
}
}
return len(packets), nil
}
func (u *TesterConn) ReloadConfig(*config.C) {}
func NewUDPStatsEmitter(_ []Conn) func() {
@@ -156,10 +131,6 @@ func (u *TesterConn) SupportsMultipleReaders() bool {
return false
}
func (u *TesterConn) BatchSize() int {
return 1
}
func (u *TesterConn) Rebind() error {
return nil
}