Merge pull request #76 from apernet/fix-enobufs

fix: engine exit with "netlink receive: recvmsg: no buffer space available" when too many packets hit NFQUEUE

README.ja.md

@@ -5,7 +5,9 @@
 [1]: https://img.shields.io/badge/License-MPL_2.0-brightgreen.svg
 [2]: LICENSE
 
-OpenGFW は、Linux 上の [GFW](https://en.wikipedia.org/wiki/Great_Firewall) の柔軟で使いやすいオープンソース実装であり、多くの点で本物より強力です。これは家庭用ルーターでできるサイバー主権です。
+OpenGFW は、あなた専用の DIY 中国のグレートファイアウォール (https://en.wikipedia.org/wiki/Great_Firewall) です。Linux 上で利用可能な柔軟で使いやすいオープンソースプログラムとして提供されています。なぜ権力者だけが楽しむのでしょうか？権力を人々に与え、検閲を民主化する時が来ました。自宅のルーターにサイバー主権のスリルをもたらし、プロのようにフィルタリングを始めましょう - あなたもビッグブラザーになることができます。
+
+Telegram グループ: https://t.me/OpGFW
 
 > [!CAUTION]
 > このプロジェクトはまだ開発の初期段階です。使用は自己責任でお願いします。
@@ -36,6 +38,7 @@ OpenGFW は、Linux 上の [GFW](https://en.wikipedia.org/wiki/Great_Firewall)
 - マルウェア対策
 - VPN/プロキシサービスの不正利用防止
 - トラフィック分析（ログのみモード）
+- 独裁的な野心を実現するのを助ける
 
 ## 使用方法
 
@@ -90,6 +93,11 @@ workers:
 式言語の構文については、[Expr 言語定義](https://expr-lang.org/docs/language-definition)を参照してください。
 
 ```yaml
+# ルールは、"action" または "log" の少なくとも一方が設定されていなければなりません。
+- name: log horny people
+  log: true
+  expr: let sni = string(tls?.req?.sni); sni contains "porn" || sni contains "hentai"
+
 - name: block v2ex http
   action: block
   expr: string(http?.req?.headers?.host) endsWith "v2ex.com"
@@ -102,8 +110,9 @@ workers:
   action: block
   expr: string(quic?.req?.sni) endsWith "v2ex.com"
 
-- name: block shadowsocks
+- name: block and log shadowsocks
   action: block
+  log: true
   expr: fet != nil && fet.yes
 
 - name: block trojan

README.md

@@ -8,8 +8,9 @@
 **[中文文档](README.zh.md)**
 **[日本語ドキュメント](README.ja.md)**
 
-OpenGFW is a flexible, easy-to-use, open source implementation of [GFW](https://en.wikipedia.org/wiki/Great_Firewall) on
-Linux that's in many ways more powerful than the real thing. It's cyber sovereignty you can have on a home router.
+OpenGFW is your very own DIY Great Firewall of China (https://en.wikipedia.org/wiki/Great_Firewall), available as a flexible, easy-to-use open source program on Linux. Why let the powers that be have all the fun? It's time to give power to the people and democratize censorship. Bring the thrill of cyber-sovereignty right into your home router and start filtering like a pro - you too can play Big Brother.
+
+Telegram group: https://t.me/OpGFW
 
 > [!CAUTION]
 > This project is still in very early stages of development. Use at your own risk.
@@ -41,6 +42,7 @@ Linux that's in many ways more powerful than the real thing. It's cyber sovereig
 - Malware protection
 - Abuse prevention for VPN/proxy services
 - Traffic analysis (log only mode)
+- Help you fulfill your dictatorial ambitions
 
 ## Usage
 
@@ -96,6 +98,11 @@ For syntax of the expression language, please refer
 to [Expr Language Definition](https://expr-lang.org/docs/language-definition).
 
 ```yaml
+# A rule must have at least one of "action" or "log" field set.
+- name: log horny people
+  log: true
+  expr: let sni = string(tls?.req?.sni); sni contains "porn" || sni contains "hentai"
+
 - name: block v2ex http
   action: block
   expr: string(http?.req?.headers?.host) endsWith "v2ex.com"
@@ -108,8 +115,9 @@ to [Expr Language Definition](https://expr-lang.org/docs/language-definition).
   action: block
   expr: string(quic?.req?.sni) endsWith "v2ex.com"
 
-- name: block shadowsocks
+- name: block and log shadowsocks
   action: block
+  log: true
   expr: fet != nil && fet.yes
 
 - name: block trojan

README.zh.md

@@ -5,8 +5,9 @@
 [1]: https://img.shields.io/badge/License-MPL_2.0-brightgreen.svg
 [2]: LICENSE
 
-OpenGFW 是一个 Linux 上灵活、易用、开源的 [GFW](https://zh.wikipedia.org/wiki/%E9%98%B2%E7%81%AB%E9%95%BF%E5%9F%8E)
-实现，并且在许多方面比真正的 GFW 更强大。可以部署在家用路由器上的网络主权。
+OpenGFW 是一个 Linux 上灵活、易用、开源的 DIY [GFW](https://zh.wikipedia.org/wiki/%E9%98%B2%E7%81%AB%E9%95%BF%E5%9F%8E) 实现，并且在许多方面比真正的 GFW 更强大。为何让那些掌权者独享乐趣？是时候把权力归还给人民，人人有墙建了。立即安装可以部署在家用路由器上的网络主权 - 你也能是老大哥。
+
+Telegram 群组： https://t.me/OpGFW
 
 > [!CAUTION]
 > 本项目仍处于早期开发阶段。测试时自行承担风险。
@@ -37,6 +38,7 @@ OpenGFW 是一个 Linux 上灵活、易用、开源的 [GFW](https://zh.wikipedi
 - 恶意软件防护
 - VPN/代理服务滥用防护
 - 流量分析 (纯日志模式)
+- 助你实现你的独裁野心
 
 ## 使用
 
@@ -91,6 +93,11 @@ workers:
 规则的语法请参考 [Expr Language Definition](https://expr-lang.org/docs/language-definition)。
 
 ```yaml
+# 每条规则必须至少包含 action 或 log 中的一个。
+- name: log horny people
+  log: true
+  expr: let sni = string(tls?.req?.sni); sni contains "porn" || sni contains "hentai"
+
 - name: block v2ex http
   action: block
   expr: string(http?.req?.headers?.host) endsWith "v2ex.com"
@@ -103,8 +110,9 @@ workers:
   action: block
   expr: string(quic?.req?.sni) endsWith "v2ex.com"
 
-- name: block shadowsocks
+- name: block and log shadowsocks
   action: block
+  log: true
   expr: fet != nil && fet.yes
 
 - name: block trojan

cmd/root.go

@@ -254,6 +254,7 @@ func runMain(cmd *cobra.Command, args []string) {
 		logger.Fatal("failed to load rules", zap.Error(err))
 	}
 	rsConfig := &ruleset.BuiltinConfig{
+		Logger:          &rulesetLogger{},
 		GeoSiteFilename: config.Ruleset.GeoSite,
 		GeoIpFilename:   config.Ruleset.GeoIp,
 	}
@@ -371,14 +372,6 @@ func (l *engineLogger) UDPStreamAction(info ruleset.StreamInfo, action ruleset.A
 		zap.Bool("noMatch", noMatch))
 }
 
-func (l *engineLogger) MatchError(info ruleset.StreamInfo, err error) {
-	logger.Error("match error",
-		zap.Int64("id", info.ID),
-		zap.String("src", info.SrcString()),
-		zap.String("dst", info.DstString()),
-		zap.Error(err))
-}
-
 func (l *engineLogger) ModifyError(info ruleset.StreamInfo, err error) {
 	logger.Error("modify error",
 		zap.Int64("id", info.ID),
@@ -408,6 +401,26 @@ func (l *engineLogger) AnalyzerErrorf(streamID int64, name string, format string
 		zap.String("msg", fmt.Sprintf(format, args...)))
 }
 
+type rulesetLogger struct{}
+
+func (l *rulesetLogger) Log(info ruleset.StreamInfo, name string) {
+	logger.Info("ruleset log",
+		zap.String("name", name),
+		zap.Int64("id", info.ID),
+		zap.String("src", info.SrcString()),
+		zap.String("dst", info.DstString()),
+		zap.Any("props", info.Props))
+}
+
+func (l *rulesetLogger) MatchError(info ruleset.StreamInfo, name string, err error) {
+	logger.Error("ruleset match error",
+		zap.String("name", name),
+		zap.Int64("id", info.ID),
+		zap.String("src", info.SrcString()),
+		zap.String("dst", info.DstString()),
+		zap.Error(err))
+}
+
 func envOrDefaultString(key, def string) string {
 	if v := os.Getenv(key); v != "" {
 		return v

engine/interface.go

@@ -41,7 +41,6 @@ type Logger interface {
 	UDPStreamPropUpdate(info ruleset.StreamInfo, close bool)
 	UDPStreamAction(info ruleset.StreamInfo, action ruleset.Action, noMatch bool)
 
-	MatchError(info ruleset.StreamInfo, err error)
 	ModifyError(info ruleset.StreamInfo, err error)
 
 	AnalyzerDebugf(streamID int64, name string, format string, args ...interface{})

engine/tcp.go

@@ -148,10 +148,7 @@ func (s *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		s.virgin = false
 		s.logger.TCPStreamPropUpdate(s.info, false)
 		// Match properties against ruleset
-		result, err := s.ruleset.Match(s.info)
-		if err != nil {
-			s.logger.MatchError(s.info, err)
-		}
+		result := s.ruleset.Match(s.info)
 		action := result.Action
 		if action != ruleset.ActionMaybe && action != ruleset.ActionModify {
 			verdict := actionToTCPVerdict(action)

engine/udp.go

@@ -201,10 +201,7 @@ func (s *udpStream) Feed(udp *layers.UDP, rev bool, uc *udpContext) {
 		s.virgin = false
 		s.logger.UDPStreamPropUpdate(s.info, false)
 		// Match properties against ruleset
-		result, err := s.ruleset.Match(s.info)
-		if err != nil {
-			s.logger.MatchError(s.info, err)
-		}
+		result := s.ruleset.Match(s.info)
 		action := result.Action
 		if action == ruleset.ActionModify {
 			// Call the modifier instance
@@ -214,6 +211,7 @@ func (s *udpStream) Feed(udp *layers.UDP, rev bool, uc *udpContext) {
 				s.logger.ModifyError(s.info, errInvalidModifier)
 				action = ruleset.ActionMaybe
 			} else {
+				var err error
 				uc.Packet, err = udpMI.Process(udp.Payload)
 				if err != nil {
 					// Modifier error, fallback to maybe

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.26.0
 	golang.org/x/crypto v0.19.0
+	golang.org/x/sys v0.17.0
 	google.golang.org/protobuf v1.31.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -43,7 +44,6 @@ require (
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
 	golang.org/x/net v0.19.0 // indirect
 	golang.org/x/sync v0.5.0 // indirect
-	golang.org/x/sys v0.17.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )

io/nfqueue.go

@@ -12,6 +12,7 @@ import (
 	"github.com/coreos/go-iptables/iptables"
 	"github.com/florianl/go-nfqueue"
 	"github.com/mdlayher/netlink"
+	"golang.org/x/sys/unix"
 )
 
 const (
@@ -138,9 +139,10 @@ func NewNFQueuePacketIO(config NFQueuePacketIOConfig) (PacketIO, error) {
 func (n *nfqueuePacketIO) Register(ctx context.Context, cb PacketCallback) error {
 	err := n.n.RegisterWithErrorFunc(ctx,
 		func(a nfqueue.Attribute) int {
-			if a.PacketID == nil || a.Ct == nil || a.Payload == nil || len(*a.Payload) < 20 {
-				// Invalid packet, ignore
-				// 20 is the minimum possible size of an IP packet
+			if ok, verdict := n.packetAttributeSanityCheck(a); !ok {
+				if a.PacketID != nil {
+					_ = n.n.SetVerdict(*a.PacketID, verdict)
+				}
 				return 0
 			}
 			p := &nfqueuePacket{
@@ -151,6 +153,12 @@ func (n *nfqueuePacketIO) Register(ctx context.Context, cb PacketCallback) error
 			return okBoolToInt(cb(p, nil))
 		},
 		func(e error) int {
+			if opErr := (*netlink.OpError)(nil); errors.As(e, &opErr) {
+				if errors.Is(opErr.Err, unix.ENOBUFS) {
+					// Kernel buffer temporarily full, ignore
+					return 0
+				}
+			}
 			return okBoolToInt(cb(nil, e))
 		})
 	if err != nil {
@@ -170,6 +178,25 @@ func (n *nfqueuePacketIO) Register(ctx context.Context, cb PacketCallback) error
 	return nil
 }
 
+func (n *nfqueuePacketIO) packetAttributeSanityCheck(a nfqueue.Attribute) (ok bool, verdict int) {
+	if a.PacketID == nil {
+		// Re-inject to NFQUEUE is actually not possible in this condition
+		return false, -1
+	}
+	if a.Payload == nil || len(*a.Payload) < 20 {
+		// 20 is the minimum possible size of an IP packet
+		return false, nfqueue.NfDrop
+	}
+	if a.Ct == nil {
+		// Multicast packets may not have a conntrack, but only appear in local mode
+		if n.local {
+			return false, nfqueue.NfAccept
+		}
+		return false, nfqueue.NfDrop
+	}
+	return true, -1
+}
+
 func (n *nfqueuePacketIO) SetVerdict(p Packet, v Verdict, newPacket []byte) error {
 	nP, ok := p.(*nfqueuePacket)
 	if !ok {

ruleset/expr.go

@@ -23,6 +23,7 @@ import (
 type ExprRule struct {
 	Name     string        `yaml:"name"`
 	Action   string        `yaml:"action"`
+	Log      bool          `yaml:"log"`
 	Modifier ModifierEntry `yaml:"modifier"`
 	Expr     string        `yaml:"expr"`
 }
@@ -45,7 +46,8 @@ func ExprRulesFromYAML(file string) ([]ExprRule, error) {
 // compiledExprRule is the internal, compiled representation of an expression rule.
 type compiledExprRule struct {
 	Name        string
-	Action      Action
+	Action      *Action // fallthrough if nil
+	Log         bool
 	ModInstance modifier.Instance
 	Program     *vm.Program
 }
@@ -55,6 +57,7 @@ var _ Ruleset = (*exprRuleset)(nil)
 type exprRuleset struct {
 	Rules      []compiledExprRule
 	Ans        []analyzer.Analyzer
+	Logger     Logger
 	GeoMatcher *geo.GeoMatcher
 }
 
@@ -62,25 +65,31 @@ func (r *exprRuleset) Analyzers(info StreamInfo) []analyzer.Analyzer {
 	return r.Ans
 }
 
-func (r *exprRuleset) Match(info StreamInfo) (MatchResult, error) {
+func (r *exprRuleset) Match(info StreamInfo) MatchResult {
 	env := streamInfoToExprEnv(info)
 	for _, rule := range r.Rules {
 		v, err := vm.Run(rule.Program, env)
 		if err != nil {
-			return MatchResult{
-				Action: ActionMaybe,
-			}, fmt.Errorf("rule %q failed to run: %w", rule.Name, err)
+			// Log the error and continue to the next rule.
+			r.Logger.MatchError(info, rule.Name, err)
+			continue
 		}
 		if vBool, ok := v.(bool); ok && vBool {
-			return MatchResult{
-				Action:      rule.Action,
-				ModInstance: rule.ModInstance,
-			}, nil
+			if rule.Log {
+				r.Logger.Log(info, rule.Name)
+			}
+			if rule.Action != nil {
+				return MatchResult{
+					Action:      *rule.Action,
+					ModInstance: rule.ModInstance,
+				}
+			}
 		}
 	}
+	// No match
 	return MatchResult{
 		Action: ActionMaybe,
-	}, nil
+	}
 }
 
 // CompileExprRules compiles a list of expression rules into a ruleset.
@@ -97,11 +106,18 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 	}
 	// Compile all rules and build a map of analyzers that are used by the rules.
 	for _, rule := range rules {
-		action, ok := actionStringToAction(rule.Action)
-		if !ok {
-			return nil, fmt.Errorf("rule %q has invalid action %q", rule.Name, rule.Action)
+		if rule.Action == "" && !rule.Log {
+			return nil, fmt.Errorf("rule %q must have at least one of action or log", rule.Name)
 		}
-		visitor := &idVisitor{Identifiers: make(map[string]bool)}
+		var action *Action
+		if rule.Action != "" {
+			a, ok := actionStringToAction(rule.Action)
+			if !ok {
+				return nil, fmt.Errorf("rule %q has invalid action %q", rule.Name, rule.Action)
+			}
+			action = &a
+		}
+		visitor := &idVisitor{Variables: make(map[string]bool), Identifiers: make(map[string]bool)}
 		patcher := &idPatcher{}
 		program, err := expr.Compile(rule.Expr,
 			func(c *conf.Config) {
@@ -118,7 +134,8 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 			return nil, fmt.Errorf("rule %q failed to patch expression: %w", rule.Name, patcher.Err)
 		}
 		for name := range visitor.Identifiers {
-			if isBuiltInAnalyzer(name) {
+			// Skip built-in analyzers & user-defined variables
+			if isBuiltInAnalyzer(name) || visitor.Variables[name] {
 				continue
 			}
 			// Check if it's one of the built-in functions, and if so,
@@ -145,9 +162,10 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 		cr := compiledExprRule{
 			Name:    rule.Name,
 			Action:  action,
+			Log:     rule.Log,
 			Program: program,
 		}
-		if action == ActionModify {
+		if action != nil && *action == ActionModify {
 			mod, ok := fullModMap[rule.Modifier.Name]
 			if !ok {
 				return nil, fmt.Errorf("rule %q uses unknown modifier %q", rule.Name, rule.Modifier.Name)
@@ -168,6 +186,7 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 	return &exprRuleset{
 		Rules:      compiledRules,
 		Ans:        depAns,
+		Logger:     config.Logger,
 		GeoMatcher: geoMatcher,
 	}, nil
 }
@@ -265,11 +284,14 @@ func modifiersToMap(mods []modifier.Modifier) map[string]modifier.Modifier {
 // idVisitor is a visitor that collects all identifiers in an expression.
 // This is for determining which analyzers are used by the expression.
 type idVisitor struct {
+	Variables   map[string]bool
 	Identifiers map[string]bool
 }
 
 func (v *idVisitor) Visit(node *ast.Node) {
-	if idNode, ok := (*node).(*ast.IdentifierNode); ok {
+	if varNode, ok := (*node).(*ast.VariableDeclaratorNode); ok {
+		v.Variables[varNode.Name] = true
+	} else if idNode, ok := (*node).(*ast.IdentifierNode); ok {
 		v.Identifiers[idNode.Value] = true
 	}
 }
@@ -284,6 +306,10 @@ func (p *idPatcher) Visit(node *ast.Node) {
 	switch (*node).(type) {
 	case *ast.CallNode:
 		callNode := (*node).(*ast.CallNode)
+		if callNode.Func == nil {
+			// Ignore invalid call nodes
+			return
+		}
 		switch callNode.Func.Name {
 		case "cidr":
 			cidrStringNode, ok := callNode.Arguments[1].(*ast.StringNode)

ruleset/interface.go

@@ -90,10 +90,17 @@ type Ruleset interface {
 	Analyzers(StreamInfo) []analyzer.Analyzer
 	// Match matches a stream against the ruleset and returns the result.
 	// It must be safe for concurrent use by multiple workers.
-	Match(StreamInfo) (MatchResult, error)
+	Match(StreamInfo) MatchResult
+}
+
+// Logger is the logging interface for the ruleset.
+type Logger interface {
+	Log(info StreamInfo, name string)
+	MatchError(info StreamInfo, name string, err error)
 }
 
 type BuiltinConfig struct {
+	Logger          Logger
 	GeoSiteFilename string
 	GeoIpFilename   string
 }