Merge pull request #135 from apernet/revert-pcap

Revert pcap

.github/workflows/release.yaml

@@ -15,13 +15,25 @@ jobs:
       matrix:
         goos: [linux]
         goarch: ["386", amd64, arm64]
+
     steps:
       - uses: actions/checkout@v4
-      - uses: wangyoucao577/go-release-action@v1
+
+      - name: Setup Go
+        uses: actions/setup-go@v5
         with:
-          github_token: ${{ secrets.GITHUB_TOKEN }}
-          goos: ${{ matrix.goos }}
-          goarch: ${{ matrix.goarch }}
-          goversion: "https://go.dev/dl/go1.22.0.linux-amd64.tar.gz"
-          binary_name: "OpenGFW"
-          extra_files: LICENSE README.md README.zh.md
+          go-version: "1.22"
+
+      - name: Build
+        env:
+          GOOS: ${{ matrix.goos }}
+          GOARCH: ${{ matrix.goarch }}
+          CGO_ENABLED: 0
+        run: |
+          mkdir -p build
+          go build -o build/OpenGFW-${GOOS}-${GOARCH} -ldflags "-s -w" .
+
+      - name: Upload
+        uses: softprops/action-gh-release@v2
+        with:
+          files: build/*

.gitignore

@@ -205,3 +205,6 @@ $RECYCLE.BIN/
 *.lnk
 
 # End of https://www.toptal.com/developers/gitignore/api/windows,macos,linux,go,goland+all,visualstudiocode
+
+# Internal tools not ready for public use yet
+tools/flowseq/

README.ja.md

@@ -8,19 +8,18 @@
 
 OpenGFW は、あなた専用の DIY 中国のグレートファイアウォール (https://en.wikipedia.org/wiki/Great_Firewall) です。Linux 上で利用可能な柔軟で使いやすいオープンソースプログラムとして提供されています。なぜ権力者だけが楽しむのでしょうか？権力を人々に与え、検閲を民主化する時が来ました。自宅のルーターにサイバー主権のスリルをもたらし、プロのようにフィルタリングを始めましょう - あなたもビッグブラザーになることができます。
 
+**ドキュメントウェブサイト: https://gfw.dev/**
+
 Telegram グループ: https://t.me/OpGFW
 
 > [!CAUTION]
-> このプロジェクトはまだ開発の初期段階です。使用は自己責任でお願いします。
-
-> [!NOTE]
-> 私たちはこのプロジェクト、特により多くのプロトコル用のアナライザーの実装を手伝ってくれるコントリビューターを探しています！！！
+> 本プロジェクトはまだ初期開発段階にあります。テスト時のリスクは自己責任でお願いします。私たちは、このプロジェクトを一緒に改善するために貢献者を探しています。
 
 ## 特徴
 
 - フル IP/TCP 再アセンブル、各種プロトコルアナライザー
-  - HTTP、TLS、QUIC、DNS、SSH、SOCKS4/5、WireGuard、その他多数
-  - Shadowsocks の「完全に暗号化されたトラフィック」の検出など (https://gfw.report/publications/usenixsecurity23/en/)
+  - HTTP、TLS、QUIC、DNS、SSH、SOCKS4/5、WireGuard、OpenVPN、その他多数
+  - Shadowsocks、VMess の「完全に暗号化されたトラフィック」の検出など (https://gfw.report/publications/usenixsecurity23/en/)
   - Trojan プロキシプロトコルの検出
   - [WIP] 機械学習に基づくトラフィック分類
 - IPv4 と IPv6 をフルサポート
@@ -40,126 +39,3 @@ Telegram グループ: https://t.me/OpGFW
 - VPN/プロキシサービスの不正利用防止
 - トラフィック分析（ログのみモード）
 - 独裁的な野心を実現するのを助ける
-
-## 使用方法
-
-### ビルド
-
-```shell
-go build
-```
-
-### 実行
-
-```shell
-export OPENGFW_LOG_LEVEL=debug
-./OpenGFW -c config.yaml rules.yaml
-```
-
-#### OpenWrt
-
-OpenGFW は OpenWrt 23.05 で動作することがテストされています（他のバージョンも動作するはずですが、検証されていません）。
-
-依存関係をインストールしてください：
-
-```shell
-# バージョン22.03以降（nftables をベースとしたファイアウォール）の場合
-opkg install kmod-nft-queue kmod-nf-conntrack-netlink
-
-# バージョン22.03以前（22.03を除く、iptablesをベースとしたファイアウォール）の場合
-opkg install kmod-ipt-nfqueue iptables-mod-nfqueue kmod-nf-conntrack-netlink
-```
-
-### 設定例
-
-```yaml
-io:
-  queueSize: 1024
-  rcvBuf: 4194304
-  sndBuf: 4194304
-  local: true # FORWARD チェーンで OpenGFW を実行したい場合は false に設定する
-  rst: false # ブロックされたTCP接続に対してRSTを送信する場合はtrueに設定してください。local=falseのみです
-
-workers:
-  count: 4
-  queueSize: 16
-  tcpMaxBufferedPagesTotal: 4096
-  tcpMaxBufferedPagesPerConn: 64
-  udpMaxStreams: 4096
-
-# 特定のローカルGeoIP / GeoSiteデータベースファイルを読み込むためのパス。
-# 設定されていない場合は、https://github.com/LoyalSoldier/v2ray-rules-dat から自動的にダウンロードされます。
-# geo:
-#   geoip: geoip.dat
-#   geosite: geosite.dat
-```
-
-### ルール例
-
-[アナライザーのプロパティ](docs/Analyzers.md)
-
-式言語の構文については、[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"
-
-- name: block v2ex https
-  action: block
-  expr: string(tls?.req?.sni) endsWith "v2ex.com"
-
-- name: block v2ex quic
-  action: block
-  expr: string(quic?.req?.sni) endsWith "v2ex.com"
-
-- name: block and log shadowsocks
-  action: block
-  log: true
-  expr: fet != nil && fet.yes
-
-- name: block trojan
-  action: block
-  expr: trojan != nil && trojan.yes
-
-- name: v2ex dns poisoning
-  action: modify
-  modifier:
-    name: dns
-    args:
-      a: "0.0.0.0"
-      aaaa: "::"
-  expr: dns != nil && dns.qr && any(dns.questions, {.name endsWith "v2ex.com"})
-
-- name: block google socks
-  action: block
-  expr: string(socks?.req?.addr) endsWith "google.com" && socks?.req?.port == 80
-
-- name: block wireguard by handshake response
-  action: drop
-  expr: wireguard?.handshake_response?.receiver_index_matched == true
-
-- name: block bilibili geosite
-  action: block
-  expr: geosite(string(tls?.req?.sni), "bilibili")
-
-- name: block CN geoip
-  action: block
-  expr: geoip(string(ip.dst), "cn")
-
-- name: block cidr
-  action: block
-  expr: cidr(string(ip.dst), "192.168.0.0/16")
-```
-
-#### サポートされるアクション
-
-- `allow`: 接続を許可し、それ以上の処理は行わない。
-- `block`: 接続をブロックし、それ以上の処理は行わない。
-- `drop`: UDP の場合、ルールのトリガーとなったパケットをドロップし、同じフローに含まれる以降のパケットの処理を継続する。TCP の場合は、`block` と同じ。
-- `modify`: UDP の場合、与えられた修飾子を使って、ルールをトリガしたパケットを修正し、同じフロー内の今後のパケットを処理し続ける。TCP の場合は、`allow` と同じ。

README.md

@@ -11,19 +11,18 @@
 
 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.
 
+**Documentation site: https://gfw.dev/**
+
 Telegram group: https://t.me/OpGFW
 
 > [!CAUTION]
-> This project is still in very early stages of development. Use at your own risk.
-
-> [!NOTE]
-> We are looking for contributors to help us with this project, especially implementing analyzers for more protocols!!!
+> This project is still in very early stages of development. Use at your own risk. We are looking for contributors to help us improve and expand the project.
 
 ## Features
 
 - Full IP/TCP reassembly, various protocol analyzers
-  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, and many more to come
-  - "Fully encrypted traffic" detection for Shadowsocks,
+  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, and many more to come
+  - "Fully encrypted traffic" detection for Shadowsocks, VMess,
     etc. (https://gfw.report/publications/usenixsecurity23/en/)
   - Trojan (proxy protocol) detection
   - [WIP] Machine learning based traffic classification
@@ -44,129 +43,3 @@ Telegram group: https://t.me/OpGFW
 - Abuse prevention for VPN/proxy services
 - Traffic analysis (log only mode)
 - Help you fulfill your dictatorial ambitions
-
-## Usage
-
-### Build
-
-```shell
-go build
-```
-
-### Run
-
-```shell
-export OPENGFW_LOG_LEVEL=debug
-./OpenGFW -c config.yaml rules.yaml
-```
-
-#### OpenWrt
-
-OpenGFW has been tested to work on OpenWrt 23.05 (other versions should also work, just not verified).
-
-Install the dependencies:
-
-```shell
-# For OpenWrt version 22.03 and later (nftables based firewall)
-opkg install kmod-nft-queue kmod-nf-conntrack-netlink
-
-# For OpenWrt versions prior to 22.03 (excluding 22.03, iptables based firewall)
-opkg install kmod-ipt-nfqueue iptables-mod-nfqueue kmod-nf-conntrack-netlink
-```
-
-### Example config
-
-```yaml
-io:
-  queueSize: 1024
-  rcvBuf: 4194304
-  sndBuf: 4194304
-  local: true # set to false if you want to run OpenGFW on FORWARD chain
-  rst: false # set to true if you want to send RST for blocked TCP connections, local=false only
-
-workers:
-  count: 4
-  queueSize: 16
-  tcpMaxBufferedPagesTotal: 4096
-  tcpMaxBufferedPagesPerConn: 64
-  udpMaxStreams: 4096
-
-# The path to load specific local geoip/geosite db files.
-# If not set, they will be automatically downloaded from https://github.com/Loyalsoldier/v2ray-rules-dat
-# geo:
-#   geoip: geoip.dat
-#   geosite: geosite.dat
-```
-
-### Example rules
-
-[Analyzer properties](docs/Analyzers.md)
-
-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"
-
-- name: block v2ex https
-  action: block
-  expr: string(tls?.req?.sni) endsWith "v2ex.com"
-
-- name: block v2ex quic
-  action: block
-  expr: string(quic?.req?.sni) endsWith "v2ex.com"
-
-- name: block and log shadowsocks
-  action: block
-  log: true
-  expr: fet != nil && fet.yes
-
-- name: block trojan
-  action: block
-  expr: trojan != nil && trojan.yes
-
-- name: v2ex dns poisoning
-  action: modify
-  modifier:
-    name: dns
-    args:
-      a: "0.0.0.0"
-      aaaa: "::"
-  expr: dns != nil && dns.qr && any(dns.questions, {.name endsWith "v2ex.com"})
-
-- name: block google socks
-  action: block
-  expr: string(socks?.req?.addr) endsWith "google.com" && socks?.req?.port == 80
-
-- name: block wireguard by handshake response
-  action: drop
-  expr: wireguard?.handshake_response?.receiver_index_matched == true
-
-- name: block bilibili geosite
-  action: block
-  expr: geosite(string(tls?.req?.sni), "bilibili")
-
-- name: block CN geoip
-  action: block
-  expr: geoip(string(ip.dst), "cn")
-
-- name: block cidr
-  action: block
-  expr: cidr(string(ip.dst), "192.168.0.0/16")
-```
-
-#### Supported actions
-
-- `allow`: Allow the connection, no further processing.
-- `block`: Block the connection, no further processing.
-- `drop`: For UDP, drop the packet that triggered the rule, continue processing future packets in the same flow. For
-  TCP, same as `block`.
-- `modify`: For UDP, modify the packet that triggered the rule using the given modifier, continue processing future
-  packets in the same flow. For TCP, same as `allow`.

README.zh.md

@@ -8,19 +8,18 @@
 
 OpenGFW 是一个 Linux 上灵活、易用、开源的 DIY [GFW](https://zh.wikipedia.org/wiki/%E9%98%B2%E7%81%AB%E9%95%BF%E5%9F%8E) 实现，并且在许多方面比真正的 GFW 更强大。为何让那些掌权者独享乐趣？是时候把权力归还给人民，人人有墙建了。立即安装可以部署在家用路由器上的网络主权 - 你也能是老大哥。
 
+**文档网站： https://gfw.dev/**
+
 Telegram 群组： https://t.me/OpGFW
 
 > [!CAUTION]
-> 本项目仍处于早期开发阶段。测试时自行承担风险。
-
-> [!NOTE]
-> 我们正在寻求贡献者一起完善本项目，尤其是实现更多协议的解析器！
+> 本项目仍处于早期开发阶段。测试时自行承担风险。我们正在寻求贡献者一起完善本项目。
 
 ## 功能
 
 - 完整的 IP/TCP 重组，各种协议解析器
-  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, 更多协议正在开发中
-  - Shadowsocks 等 "全加密流量" 检测 (https://gfw.report/publications/usenixsecurity23/zh/)
+  - HTTP, TLS, QUIC, DNS, SSH, SOCKS4/5, WireGuard, OpenVPN, 更多协议正在开发中
+  - Shadowsocks, VMess 等 "全加密流量" 检测 (https://gfw.report/publications/usenixsecurity23/zh/)
   - Trojan 协议检测
   - [开发中] 基于机器学习的流量分类
 - 同等支持 IPv4 和 IPv6
@@ -40,126 +39,3 @@ Telegram 群组： https://t.me/OpGFW
 - VPN/代理服务滥用防护
 - 流量分析 (纯日志模式)
 - 助你实现你的独裁野心
-
-## 使用
-
-### 构建
-
-```shell
-go build
-```
-
-### 运行
-
-```shell
-export OPENGFW_LOG_LEVEL=debug
-./OpenGFW -c config.yaml rules.yaml
-```
-
-#### OpenWrt
-
-OpenGFW 在 OpenWrt 23.05 上测试可用（其他版本应该也可以，暂时未经验证）。
-
-安装依赖：
-
-```shell
-# 对于 22.03 或者之后的版本（基于 nftables 的防火墙）
-opkg install kmod-nft-queue kmod-nf-conntrack-netlink
-
-# 对于 22.03 之前的版本（不包括 22.03， 基于 iptables 的防火墙）
-opkg install kmod-ipt-nfqueue iptables-mod-nfqueue kmod-nf-conntrack-netlink
-```
-
-### 样例配置
-
-```yaml
-io:
-  queueSize: 1024
-  rcvBuf: 4194304
-  sndBuf: 4194304
-  local: true # 如果需要在 FORWARD 链上运行 OpenGFW，请设置为 false
-  rst: false # 是否对要阻断的 TCP 连接发送 RST。仅在 local=false 时有效
-
-workers:
-  count: 4
-  queueSize: 16
-  tcpMaxBufferedPagesTotal: 4096
-  tcpMaxBufferedPagesPerConn: 64
-  udpMaxStreams: 4096
-
-# 指定的 geoip/geosite 档案路径
-# 如果未设置，将自动从 https://github.com/Loyalsoldier/v2ray-rules-dat 下载
-# geo:
-#   geoip: geoip.dat
-#   geosite: geosite.dat
-```
-
-### 样例规则
-
-[解析器属性](docs/Analyzers.md)
-
-规则的语法请参考 [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"
-
-- name: block v2ex https
-  action: block
-  expr: string(tls?.req?.sni) endsWith "v2ex.com"
-
-- name: block v2ex quic
-  action: block
-  expr: string(quic?.req?.sni) endsWith "v2ex.com"
-
-- name: block and log shadowsocks
-  action: block
-  log: true
-  expr: fet != nil && fet.yes
-
-- name: block trojan
-  action: block
-  expr: trojan != nil && trojan.yes
-
-- name: v2ex dns poisoning
-  action: modify
-  modifier:
-    name: dns
-    args:
-      a: "0.0.0.0"
-      aaaa: "::"
-  expr: dns != nil && dns.qr && any(dns.questions, {.name endsWith "v2ex.com"})
-
-- name: block google socks
-  action: block
-  expr: string(socks?.req?.addr) endsWith "google.com" && socks?.req?.port == 80
-
-- name: block wireguard by handshake response
-  action: drop
-  expr: wireguard?.handshake_response?.receiver_index_matched == true
-
-- name: block bilibili geosite
-  action: block
-  expr: geosite(string(tls?.req?.sni), "bilibili")
-
-- name: block CN geoip
-  action: block
-  expr: geoip(string(ip.dst), "cn")
-
-- name: block cidr
-  action: block
-  expr: cidr(string(ip.dst), "192.168.0.0/16")
-```
-
-#### 支持的 action
-
-- `allow`: 放行连接，不再处理后续的包。
-- `block`: 阻断连接，不再处理后续的包。
-- `drop`: 对于 UDP，丢弃触发规则的包，但继续处理同一流中的后续包。对于 TCP，效果同 `block`。
-- `modify`: 对于 UDP，用指定的修改器修改触发规则的包，然后继续处理同一流中的后续包。对于 TCP，效果同 `allow`。

analyzer/tcp/trojan.go

@@ -13,10 +13,11 @@ var ccsPattern = []byte{20, 3, 3, 0, 1, 1}
 
 // TrojanAnalyzer uses length-based heuristics to detect Trojan traffic based on
 // its "TLS-in-TLS" nature. The heuristics are trained using a decision tree with
-// about 2000 samples. This is highly experimental and is known to have significant
-// false positives (about 9% false positives & 3% false negatives).
-// We do NOT recommend directly blocking all positive connections, as this is likely
-// to break many normal TLS connections.
+// about 20k Trojan samples and 30k non-Trojan samples. The tree is then converted
+// to code using a custom tool and inlined here (isTrojanSeq function).
+// Accuracy: 1% false positive rate, 10% false negative rate.
+// We do NOT recommend directly blocking all positive connections, as this may
+// break legitimate TLS connections.
 type TrojanAnalyzer struct{}
 
 func (a *TrojanAnalyzer) Name() string {
@@ -36,7 +37,7 @@ type trojanStream struct {
 	first    bool
 	count    bool
 	rev      bool
-	seq      [3]int
+	seq      [4]int
 	seqIndex int
 }
 
@@ -73,16 +74,12 @@ func (s *trojanStream) Feed(rev, start, end bool, skip int, data []byte) (u *ana
 		} else {
 			// Different direction, bump the index
 			s.seqIndex += 1
-			if s.seqIndex == 3 {
-				// Time to evaluate
-				yes := s.seq[0] >= 180 &&
-					s.seq[1] <= 11000 &&
-					s.seq[2] >= 40
+			if s.seqIndex == 4 {
 				return &analyzer.PropUpdate{
 					Type: analyzer.PropUpdateReplace,
 					M: analyzer.PropMap{
 						"seq": s.seq,
-						"yes": yes,
+						"yes": isTrojanSeq(s.seq),
 					},
 				}, true
 			}
@@ -97,3 +94,424 @@ func (s *trojanStream) Feed(rev, start, end bool, skip int, data []byte) (u *ana
 func (s *trojanStream) Close(limited bool) *analyzer.PropUpdate {
 	return nil
 }
+
+func isTrojanSeq(seq [4]int) bool {
+	length1 := seq[0]
+	length2 := seq[1]
+	length3 := seq[2]
+	length4 := seq[3]
+
+	if length2 <= 2431 {
+		if length2 <= 157 {
+			if length1 <= 156 {
+				if length3 <= 108 {
+					return false
+				} else {
+					return false
+				}
+			} else {
+				if length1 <= 892 {
+					if length3 <= 40 {
+						return false
+					} else {
+						if length3 <= 788 {
+							if length4 <= 185 {
+								if length1 <= 411 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								if length2 <= 112 {
+									return false
+								} else {
+									return true
+								}
+							}
+						} else {
+							if length3 <= 1346 {
+								if length1 <= 418 {
+									return false
+								} else {
+									return true
+								}
+							} else {
+								return false
+							}
+						}
+					}
+				} else {
+					if length2 <= 120 {
+						if length2 <= 63 {
+							return false
+						} else {
+							if length4 <= 653 {
+								return false
+							} else {
+								return false
+							}
+						}
+					} else {
+						return false
+					}
+				}
+			}
+		} else {
+			if length1 <= 206 {
+				if length1 <= 185 {
+					if length1 <= 171 {
+						return false
+					} else {
+						if length4 <= 211 {
+							return false
+						} else {
+							return false
+						}
+					}
+				} else {
+					if length2 <= 251 {
+						return true
+					} else {
+						return false
+					}
+				}
+			} else {
+				if length2 <= 286 {
+					if length1 <= 1123 {
+						if length3 <= 70 {
+							return false
+						} else {
+							if length1 <= 659 {
+								if length3 <= 370 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								if length4 <= 272 {
+									return false
+								} else {
+									return true
+								}
+							}
+						}
+					} else {
+						if length4 <= 537 {
+							if length2 <= 276 {
+								if length3 <= 1877 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								return false
+							}
+						} else {
+							if length1 <= 1466 {
+								if length1 <= 1435 {
+									return false
+								} else {
+									return true
+								}
+							} else {
+								if length2 <= 193 {
+									return false
+								} else {
+									return false
+								}
+							}
+						}
+					}
+				} else {
+					if length1 <= 284 {
+						if length1 <= 277 {
+							if length2 <= 726 {
+								return false
+							} else {
+								if length2 <= 768 {
+									return true
+								} else {
+									return false
+								}
+							}
+						} else {
+							if length2 <= 782 {
+								if length4 <= 783 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								return false
+							}
+						}
+					} else {
+						if length2 <= 492 {
+							if length2 <= 396 {
+								if length2 <= 322 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length4 <= 971 {
+									return false
+								} else {
+									return true
+								}
+							}
+						} else {
+							if length2 <= 2128 {
+								if length2 <= 1418 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length3 <= 103 {
+									return false
+								} else {
+									return false
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	} else {
+		if length2 <= 6232 {
+			if length3 <= 85 {
+				if length2 <= 3599 {
+					return false
+				} else {
+					if length1 <= 613 {
+						return false
+					} else {
+						return false
+					}
+				}
+			} else {
+				if length3 <= 220 {
+					if length4 <= 1173 {
+						if length1 <= 874 {
+							if length4 <= 337 {
+								if length4 <= 68 {
+									return true
+								} else {
+									return true
+								}
+							} else {
+								if length1 <= 667 {
+									return true
+								} else {
+									return true
+								}
+							}
+						} else {
+							if length3 <= 108 {
+								if length1 <= 1930 {
+									return true
+								} else {
+									return true
+								}
+							} else {
+								if length2 <= 5383 {
+									return false
+								} else {
+									return true
+								}
+							}
+						}
+					} else {
+						return false
+					}
+				} else {
+					if length1 <= 664 {
+						if length3 <= 411 {
+							if length3 <= 383 {
+								if length4 <= 346 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								if length1 <= 445 {
+									return true
+								} else {
+									return false
+								}
+							}
+						} else {
+							if length2 <= 3708 {
+								if length4 <= 307 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								if length2 <= 4656 {
+									return false
+								} else {
+									return false
+								}
+							}
+						}
+					} else {
+						if length1 <= 1055 {
+							if length3 <= 580 {
+								if length1 <= 724 {
+									return true
+								} else {
+									return false
+								}
+							} else {
+								if length1 <= 678 {
+									return false
+								} else {
+									return true
+								}
+							}
+						} else {
+							if length2 <= 5352 {
+								if length3 <= 1586 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length4 <= 2173 {
+									return true
+								} else {
+									return false
+								}
+							}
+						}
+					}
+				}
+			}
+		} else {
+			if length2 <= 9408 {
+				if length1 <= 670 {
+					if length4 <= 76 {
+						if length3 <= 175 {
+							return true
+						} else {
+							return true
+						}
+					} else {
+						if length2 <= 9072 {
+							if length3 <= 314 {
+								if length3 <= 179 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length4 <= 708 {
+									return false
+								} else {
+									return false
+								}
+							}
+						} else {
+							return true
+						}
+					}
+				} else {
+					if length1 <= 795 {
+						if length2 <= 6334 {
+							if length2 <= 6288 {
+								return true
+							} else {
+								return false
+							}
+						} else {
+							if length4 <= 6404 {
+								if length2 <= 8194 {
+									return true
+								} else {
+									return true
+								}
+							} else {
+								if length2 <= 8924 {
+									return false
+								} else {
+									return true
+								}
+							}
+						}
+					} else {
+						if length3 <= 732 {
+							if length1 <= 1397 {
+								if length3 <= 179 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length1 <= 1976 {
+									return false
+								} else {
+									return false
+								}
+							}
+						} else {
+							if length1 <= 2840 {
+								if length1 <= 2591 {
+									return false
+								} else {
+									return true
+								}
+							} else {
+								return false
+							}
+						}
+					}
+				}
+			} else {
+				if length4 <= 30 {
+					return false
+				} else {
+					if length2 <= 13314 {
+						if length4 <= 1786 {
+							if length2 <= 13018 {
+								if length4 <= 869 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								return true
+							}
+						} else {
+							if length3 <= 775 {
+								return false
+							} else {
+								return false
+							}
+						}
+					} else {
+						if length4 <= 73 {
+							return false
+						} else {
+							if length3 <= 640 {
+								if length3 <= 237 {
+									return false
+								} else {
+									return false
+								}
+							} else {
+								if length2 <= 43804 {
+									return false
+								} else {
+									return false
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+}

analyzer/udp/openvpn.go

@@ -0,0 +1,384 @@
+package udp
+
+import (
+	"github.com/apernet/OpenGFW/analyzer"
+	"github.com/apernet/OpenGFW/analyzer/utils"
+)
+
+var (
+	_ analyzer.UDPAnalyzer = (*OpenVPNAnalyzer)(nil)
+	_ analyzer.TCPAnalyzer = (*OpenVPNAnalyzer)(nil)
+)
+
+var (
+	_ analyzer.UDPStream = (*openvpnUDPStream)(nil)
+	_ analyzer.TCPStream = (*openvpnTCPStream)(nil)
+)
+
+// Ref paper:
+// https://www.usenix.org/system/files/sec22fall_xue-diwen.pdf
+
+// OpenVPN Opcodes definitions from:
+// https://github.com/OpenVPN/openvpn/blob/master/src/openvpn/ssl_pkt.h
+const (
+	OpenVPNControlHardResetClientV1 = 1
+	OpenVPNControlHardResetServerV1 = 2
+	OpenVPNControlSoftResetV1       = 3
+	OpenVPNControlV1                = 4
+	OpenVPNAckV1                    = 5
+	OpenVPNDataV1                   = 6
+	OpenVPNControlHardResetClientV2 = 7
+	OpenVPNControlHardResetServerV2 = 8
+	OpenVPNDataV2                   = 9
+	OpenVPNControlHardResetClientV3 = 10
+	OpenVPNControlWkcV1             = 11
+)
+
+const (
+	OpenVPNMinPktLen          = 6
+	OpenVPNTCPPktDefaultLimit = 256
+	OpenVPNUDPPktDefaultLimit = 256
+)
+
+type OpenVPNAnalyzer struct{}
+
+func (a *OpenVPNAnalyzer) Name() string {
+	return "openvpn"
+}
+
+func (a *OpenVPNAnalyzer) Limit() int {
+	return 0
+}
+
+func (a *OpenVPNAnalyzer) NewUDP(info analyzer.UDPInfo, logger analyzer.Logger) analyzer.UDPStream {
+	return newOpenVPNUDPStream(logger)
+}
+
+func (a *OpenVPNAnalyzer) NewTCP(info analyzer.TCPInfo, logger analyzer.Logger) analyzer.TCPStream {
+	return newOpenVPNTCPStream(logger)
+}
+
+type openvpnPkt struct {
+	pktLen uint16 // 16 bits, TCP proto only
+	opcode byte   // 5 bits
+	_keyId byte   // 3 bits, not used
+
+	// We don't care about the rest of the packet
+	// payload []byte
+}
+
+type openvpnStream struct {
+	logger analyzer.Logger
+
+	reqUpdated bool
+	reqLSM     *utils.LinearStateMachine
+	reqDone    bool
+
+	respUpdated bool
+	respLSM     *utils.LinearStateMachine
+	respDone    bool
+
+	rxPktCnt int
+	txPktCnt int
+	pktLimit int
+
+	reqPktParse  func() (*openvpnPkt, utils.LSMAction)
+	respPktParse func() (*openvpnPkt, utils.LSMAction)
+
+	lastOpcode byte
+}
+
+func (o *openvpnStream) parseCtlHardResetClient() utils.LSMAction {
+	pkt, action := o.reqPktParse()
+	if action != utils.LSMActionNext {
+		return action
+	}
+
+	if pkt.opcode != OpenVPNControlHardResetClientV1 &&
+		pkt.opcode != OpenVPNControlHardResetClientV2 &&
+		pkt.opcode != OpenVPNControlHardResetClientV3 {
+		return utils.LSMActionCancel
+	}
+	o.lastOpcode = pkt.opcode
+
+	return utils.LSMActionNext
+}
+
+func (o *openvpnStream) parseCtlHardResetServer() utils.LSMAction {
+	if o.lastOpcode != OpenVPNControlHardResetClientV1 &&
+		o.lastOpcode != OpenVPNControlHardResetClientV2 &&
+		o.lastOpcode != OpenVPNControlHardResetClientV3 {
+		return utils.LSMActionCancel
+	}
+
+	pkt, action := o.respPktParse()
+	if action != utils.LSMActionNext {
+		return action
+	}
+
+	if pkt.opcode != OpenVPNControlHardResetServerV1 &&
+		pkt.opcode != OpenVPNControlHardResetServerV2 {
+		return utils.LSMActionCancel
+	}
+	o.lastOpcode = pkt.opcode
+
+	return utils.LSMActionNext
+}
+
+func (o *openvpnStream) parseReq() utils.LSMAction {
+	pkt, action := o.reqPktParse()
+	if action != utils.LSMActionNext {
+		return action
+	}
+
+	if pkt.opcode != OpenVPNControlSoftResetV1 &&
+		pkt.opcode != OpenVPNControlV1 &&
+		pkt.opcode != OpenVPNAckV1 &&
+		pkt.opcode != OpenVPNDataV1 &&
+		pkt.opcode != OpenVPNDataV2 &&
+		pkt.opcode != OpenVPNControlWkcV1 {
+		return utils.LSMActionCancel
+	}
+
+	o.txPktCnt += 1
+	o.reqUpdated = true
+
+	return utils.LSMActionPause
+}
+
+func (o *openvpnStream) parseResp() utils.LSMAction {
+	pkt, action := o.respPktParse()
+	if action != utils.LSMActionNext {
+		return action
+	}
+
+	if pkt.opcode != OpenVPNControlSoftResetV1 &&
+		pkt.opcode != OpenVPNControlV1 &&
+		pkt.opcode != OpenVPNAckV1 &&
+		pkt.opcode != OpenVPNDataV1 &&
+		pkt.opcode != OpenVPNDataV2 &&
+		pkt.opcode != OpenVPNControlWkcV1 {
+		return utils.LSMActionCancel
+	}
+
+	o.rxPktCnt += 1
+	o.respUpdated = true
+
+	return utils.LSMActionPause
+}
+
+type openvpnUDPStream struct {
+	openvpnStream
+	curPkt []byte
+	// We don't introduce `invalidCount` here to decrease the false positive rate
+	// invalidCount int
+}
+
+func newOpenVPNUDPStream(logger analyzer.Logger) *openvpnUDPStream {
+	s := &openvpnUDPStream{
+		openvpnStream: openvpnStream{
+			logger:   logger,
+			pktLimit: OpenVPNUDPPktDefaultLimit,
+		},
+	}
+	s.respPktParse = s.parsePkt
+	s.reqPktParse = s.parsePkt
+	s.reqLSM = utils.NewLinearStateMachine(
+		s.parseCtlHardResetClient,
+		s.parseReq,
+	)
+	s.respLSM = utils.NewLinearStateMachine(
+		s.parseCtlHardResetServer,
+		s.parseResp,
+	)
+	return s
+}
+
+func (o *openvpnUDPStream) Feed(rev bool, data []byte) (u *analyzer.PropUpdate, d bool) {
+	if len(data) == 0 {
+		return nil, false
+	}
+	var update *analyzer.PropUpdate
+	var cancelled bool
+	o.curPkt = data
+	if rev {
+		o.respUpdated = false
+		cancelled, o.respDone = o.respLSM.Run()
+		if o.respUpdated {
+			update = &analyzer.PropUpdate{
+				Type: analyzer.PropUpdateReplace,
+				M:    analyzer.PropMap{"rx_pkt_cnt": o.rxPktCnt, "tx_pkt_cnt": o.txPktCnt},
+			}
+			o.respUpdated = false
+		}
+	} else {
+		o.reqUpdated = false
+		cancelled, o.reqDone = o.reqLSM.Run()
+		if o.reqUpdated {
+			update = &analyzer.PropUpdate{
+				Type: analyzer.PropUpdateReplace,
+				M:    analyzer.PropMap{"rx_pkt_cnt": o.rxPktCnt, "tx_pkt_cnt": o.txPktCnt},
+			}
+			o.reqUpdated = false
+		}
+	}
+
+	return update, cancelled || (o.reqDone && o.respDone) || o.rxPktCnt+o.txPktCnt > o.pktLimit
+}
+
+func (o *openvpnUDPStream) Close(limited bool) *analyzer.PropUpdate {
+	return nil
+}
+
+// Parse OpenVPN UDP packet.
+func (o *openvpnUDPStream) parsePkt() (p *openvpnPkt, action utils.LSMAction) {
+	if o.curPkt == nil {
+		return nil, utils.LSMActionPause
+	}
+
+	if !OpenVPNCheckForValidOpcode(o.curPkt[0] >> 3) {
+		return nil, utils.LSMActionCancel
+	}
+
+	// Parse packet header
+	p = &openvpnPkt{}
+	p.opcode = o.curPkt[0] >> 3
+	p._keyId = o.curPkt[0] & 0x07
+
+	o.curPkt = nil
+	return p, utils.LSMActionNext
+}
+
+type openvpnTCPStream struct {
+	openvpnStream
+	reqBuf  *utils.ByteBuffer
+	respBuf *utils.ByteBuffer
+}
+
+func newOpenVPNTCPStream(logger analyzer.Logger) *openvpnTCPStream {
+	s := &openvpnTCPStream{
+		openvpnStream: openvpnStream{
+			logger:   logger,
+			pktLimit: OpenVPNTCPPktDefaultLimit,
+		},
+		reqBuf:  &utils.ByteBuffer{},
+		respBuf: &utils.ByteBuffer{},
+	}
+	s.respPktParse = func() (*openvpnPkt, utils.LSMAction) {
+		return s.parsePkt(true)
+	}
+	s.reqPktParse = func() (*openvpnPkt, utils.LSMAction) {
+		return s.parsePkt(false)
+	}
+	s.reqLSM = utils.NewLinearStateMachine(
+		s.parseCtlHardResetClient,
+		s.parseReq,
+	)
+	s.respLSM = utils.NewLinearStateMachine(
+		s.parseCtlHardResetServer,
+		s.parseResp,
+	)
+	return s
+}
+
+func (o *openvpnTCPStream) Feed(rev, start, end bool, skip int, data []byte) (u *analyzer.PropUpdate, d bool) {
+	if skip != 0 {
+		return nil, true
+	}
+	if len(data) == 0 {
+		return nil, false
+	}
+	var update *analyzer.PropUpdate
+	var cancelled bool
+	if rev {
+		o.respBuf.Append(data)
+		o.respUpdated = false
+		cancelled, o.respDone = o.respLSM.Run()
+		if o.respUpdated {
+			update = &analyzer.PropUpdate{
+				Type: analyzer.PropUpdateReplace,
+				M:    analyzer.PropMap{"rx_pkt_cnt": o.rxPktCnt, "tx_pkt_cnt": o.txPktCnt},
+			}
+			o.respUpdated = false
+		}
+	} else {
+		o.reqBuf.Append(data)
+		o.reqUpdated = false
+		cancelled, o.reqDone = o.reqLSM.Run()
+		if o.reqUpdated {
+			update = &analyzer.PropUpdate{
+				Type: analyzer.PropUpdateMerge,
+				M:    analyzer.PropMap{"rx_pkt_cnt": o.rxPktCnt, "tx_pkt_cnt": o.txPktCnt},
+			}
+			o.reqUpdated = false
+		}
+	}
+
+	return update, cancelled || (o.reqDone && o.respDone) || o.rxPktCnt+o.txPktCnt > o.pktLimit
+}
+
+func (o *openvpnTCPStream) Close(limited bool) *analyzer.PropUpdate {
+	o.reqBuf.Reset()
+	o.respBuf.Reset()
+	return nil
+}
+
+// Parse OpenVPN TCP packet.
+func (o *openvpnTCPStream) parsePkt(rev bool) (p *openvpnPkt, action utils.LSMAction) {
+	var buffer *utils.ByteBuffer
+	if rev {
+		buffer = o.respBuf
+	} else {
+		buffer = o.reqBuf
+	}
+
+	// Parse packet length
+	pktLen, ok := buffer.GetUint16(false, false)
+	if !ok {
+		return nil, utils.LSMActionPause
+	}
+
+	if pktLen < OpenVPNMinPktLen {
+		return nil, utils.LSMActionCancel
+	}
+
+	pktOp, ok := buffer.Get(3, false)
+	if !ok {
+		return nil, utils.LSMActionPause
+	}
+	if !OpenVPNCheckForValidOpcode(pktOp[2] >> 3) {
+		return nil, utils.LSMActionCancel
+	}
+
+	pkt, ok := buffer.Get(int(pktLen)+2, true)
+	if !ok {
+		return nil, utils.LSMActionPause
+	}
+	pkt = pkt[2:]
+
+	// Parse packet header
+	p = &openvpnPkt{}
+	p.pktLen = pktLen
+	p.opcode = pkt[0] >> 3
+	p._keyId = pkt[0] & 0x07
+
+	return p, utils.LSMActionNext
+}
+
+func OpenVPNCheckForValidOpcode(opcode byte) bool {
+	switch opcode {
+	case OpenVPNControlHardResetClientV1,
+		OpenVPNControlHardResetServerV1,
+		OpenVPNControlSoftResetV1,
+		OpenVPNControlV1,
+		OpenVPNAckV1,
+		OpenVPNDataV1,
+		OpenVPNControlHardResetClientV2,
+		OpenVPNControlHardResetServerV2,
+		OpenVPNDataV2,
+		OpenVPNControlHardResetClientV3,
+		OpenVPNControlWkcV1:
+		return true
+	}
+	return false
+}

cmd/root.go

@@ -7,6 +7,7 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/apernet/OpenGFW/analyzer"
 	"github.com/apernet/OpenGFW/analyzer/tcp"
@@ -16,6 +17,7 @@ import (
 	"github.com/apernet/OpenGFW/modifier"
 	modUDP "github.com/apernet/OpenGFW/modifier/udp"
 	"github.com/apernet/OpenGFW/ruleset"
+	"github.com/apernet/OpenGFW/ruleset/builtins/geo"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -92,6 +94,7 @@ var analyzers = []analyzer.Analyzer{
 	&tcp.TLSAnalyzer{},
 	&tcp.TrojanAnalyzer{},
 	&udp.DNSAnalyzer{},
+	&udp.OpenVPNAnalyzer{},
 	&udp.QUICAnalyzer{},
 	&udp.WireGuardAnalyzer{},
 }
@@ -175,11 +178,12 @@ type cliConfigIO struct {
 }
 
 type cliConfigWorkers struct {
-	Count                      int `mapstructure:"count"`
-	QueueSize                  int `mapstructure:"queueSize"`
-	TCPMaxBufferedPagesTotal   int `mapstructure:"tcpMaxBufferedPagesTotal"`
-	TCPMaxBufferedPagesPerConn int `mapstructure:"tcpMaxBufferedPagesPerConn"`
-	UDPMaxStreams              int `mapstructure:"udpMaxStreams"`
+	Count                      int           `mapstructure:"count"`
+	QueueSize                  int           `mapstructure:"queueSize"`
+	TCPMaxBufferedPagesTotal   int           `mapstructure:"tcpMaxBufferedPagesTotal"`
+	TCPMaxBufferedPagesPerConn int           `mapstructure:"tcpMaxBufferedPagesPerConn"`
+	TCPTimeout                 time.Duration `mapstructure:"tcpTimeout"`
+	UDPMaxStreams              int           `mapstructure:"udpMaxStreams"`
 }
 
 type cliConfigRuleset struct {
@@ -203,7 +207,7 @@ func (c *cliConfig) fillIO(config *engine.Config) error {
 	if err != nil {
 		return configError{Field: "io", Err: err}
 	}
-	config.IOs = []io.PacketIO{nfio}
+	config.IO = nfio
 	return nil
 }
 
@@ -212,6 +216,7 @@ func (c *cliConfig) fillWorkers(config *engine.Config) error {
 	config.WorkerQueueSize = c.Workers.QueueSize
 	config.WorkerTCPMaxBufferedPagesTotal = c.Workers.TCPMaxBufferedPagesTotal
 	config.WorkerTCPMaxBufferedPagesPerConn = c.Workers.TCPMaxBufferedPagesPerConn
+	config.WorkerTCPTimeout = c.Workers.TCPTimeout
 	config.WorkerUDPMaxStreams = c.Workers.UDPMaxStreams
 	return nil
 }
@@ -246,12 +251,7 @@ func runMain(cmd *cobra.Command, args []string) {
 	if err != nil {
 		logger.Fatal("failed to parse config", zap.Error(err))
 	}
-	defer func() {
-		// Make sure to close all IOs on exit
-		for _, i := range engineConfig.IOs {
-			_ = i.Close()
-		}
-	}()
+	defer engineConfig.IO.Close() // Make sure to close IO on exit
 
 	// Ruleset
 	rawRs, err := ruleset.ExprRulesFromYAML(args[0])
@@ -259,9 +259,9 @@ 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,
+		Logger:               &rulesetLogger{},
+		GeoMatcher:           geo.NewGeoMatcher(config.Ruleset.GeoSite, config.Ruleset.GeoIp),
+		ProtectedDialContext: engineConfig.IO.ProtectedDialContext,
 	}
 	rs, err := ruleset.CompileExprRules(rawRs, analyzers, modifiers, rsConfig)
 	if err != nil {
@@ -343,12 +343,26 @@ func (l *engineLogger) TCPStreamPropUpdate(info ruleset.StreamInfo, close bool)
 }
 
 func (l *engineLogger) TCPStreamAction(info ruleset.StreamInfo, action ruleset.Action, noMatch bool) {
-	logger.Info("TCP stream action",
-		zap.Int64("id", info.ID),
-		zap.String("src", info.SrcString()),
-		zap.String("dst", info.DstString()),
-		zap.String("action", action.String()),
-		zap.Bool("noMatch", noMatch))
+	if noMatch {
+		logger.Debug("TCP stream no match",
+			zap.Int64("id", info.ID),
+			zap.String("src", info.SrcString()),
+			zap.String("dst", info.DstString()),
+			zap.String("action", action.String()))
+	} else {
+		logger.Info("TCP stream action",
+			zap.Int64("id", info.ID),
+			zap.String("src", info.SrcString()),
+			zap.String("dst", info.DstString()),
+			zap.String("action", action.String()))
+	}
+}
+
+func (l *engineLogger) TCPFlush(workerID, flushed, closed int) {
+	logger.Debug("TCP flush",
+		zap.Int("workerID", workerID),
+		zap.Int("flushed", flushed),
+		zap.Int("closed", closed))
 }
 
 func (l *engineLogger) UDPStreamNew(workerID int, info ruleset.StreamInfo) {
@@ -369,12 +383,19 @@ func (l *engineLogger) UDPStreamPropUpdate(info ruleset.StreamInfo, close bool)
 }
 
 func (l *engineLogger) UDPStreamAction(info ruleset.StreamInfo, action ruleset.Action, noMatch bool) {
-	logger.Info("UDP stream action",
-		zap.Int64("id", info.ID),
-		zap.String("src", info.SrcString()),
-		zap.String("dst", info.DstString()),
-		zap.String("action", action.String()),
-		zap.Bool("noMatch", noMatch))
+	if noMatch {
+		logger.Debug("UDP stream no match",
+			zap.Int64("id", info.ID),
+			zap.String("src", info.SrcString()),
+			zap.String("dst", info.DstString()),
+			zap.String("action", action.String()))
+	} else {
+		logger.Info("UDP stream action",
+			zap.Int64("id", info.ID),
+			zap.String("src", info.SrcString()),
+			zap.String("dst", info.DstString()),
+			zap.String("action", action.String()))
+	}
 }
 
 func (l *engineLogger) ModifyError(info ruleset.StreamInfo, err error) {

docs/Analyzers.md

@@ -1,415 +0,0 @@
-# Analyzers
-
-Analyzers are one of the main components of OpenGFW. Their job is to analyze a connection, see if it's a protocol they
-support, and if so, extract information from that connection and provide properties for the rule engine to match against
-user-provided rules. OpenGFW will automatically analyze which analyzers are referenced in the given rules and enable
-only those that are needed.
-
-This document lists the properties provided by each analyzer that can be used by rules.
-
-## DNS (TCP & UDP)
-
-For queries:
-
-```json
-{
-  "dns": {
-    "aa": false,
-    "id": 41953,
-    "opcode": 0,
-    "qr": false,
-    "questions": [
-      {
-        "class": 1,
-        "name": "www.google.com",
-        "type": 1
-      }
-    ],
-    "ra": false,
-    "rcode": 0,
-    "rd": true,
-    "tc": false,
-    "z": 0
-  }
-}
-```
-
-For responses:
-
-```json
-{
-  "dns": {
-    "aa": false,
-    "answers": [
-      {
-        "a": "142.251.32.36",
-        "class": 1,
-        "name": "www.google.com",
-        "ttl": 255,
-        "type": 1
-      }
-    ],
-    "id": 41953,
-    "opcode": 0,
-    "qr": true,
-    "questions": [
-      {
-        "class": 1,
-        "name": "www.google.com",
-        "type": 1
-      }
-    ],
-    "ra": true,
-    "rcode": 0,
-    "rd": true,
-    "tc": false,
-    "z": 0
-  }
-}
-```
-
-Example for blocking DNS queries for `www.google.com`:
-
-```yaml
-- name: Block Google DNS
-  action: drop
-  expr: dns != nil && !dns.qr && any(dns.questions, {.name == "www.google.com"})
-```
-
-## FET (Fully Encrypted Traffic)
-
-Check https://www.usenix.org/system/files/usenixsecurity23-wu-mingshi.pdf for more information.
-
-```json
-{
-  "fet": {
-    "ex1": 3.7560976,
-    "ex2": true,
-    "ex3": 0.9512195,
-    "ex4": 39,
-    "ex5": false,
-    "yes": false
-  }
-}
-```
-
-Example for blocking fully encrypted traffic:
-
-```yaml
-- name: Block suspicious proxy traffic
-  action: block
-  expr: fet != nil && fet.yes
-```
-
-## HTTP
-
-```json
-{
-  "http": {
-    "req": {
-      "headers": {
-        "accept": "*/*",
-        "host": "ipinfo.io",
-        "user-agent": "curl/7.81.0"
-      },
-      "method": "GET",
-      "path": "/",
-      "version": "HTTP/1.1"
-    },
-    "resp": {
-      "headers": {
-        "access-control-allow-origin": "*",
-        "content-length": "333",
-        "content-type": "application/json; charset=utf-8",
-        "date": "Wed, 24 Jan 2024 05:41:44 GMT",
-        "referrer-policy": "strict-origin-when-cross-origin",
-        "server": "nginx/1.24.0",
-        "strict-transport-security": "max-age=2592000; includeSubDomains",
-        "via": "1.1 google",
-        "x-content-type-options": "nosniff",
-        "x-envoy-upstream-service-time": "2",
-        "x-frame-options": "SAMEORIGIN",
-        "x-xss-protection": "1; mode=block"
-      },
-      "status": 200,
-      "version": "HTTP/1.1"
-    }
-  }
-}
-```
-
-Example for blocking HTTP requests to `ipinfo.io`:
-
-```yaml
-- name: Block ipinfo.io HTTP
-  action: block
-  expr: http != nil && http.req != nil && http.req.headers != nil && http.req.headers.host == "ipinfo.io"
-```
-
-## SSH
-
-```json
-{
-  "ssh": {
-    "server": {
-      "comments": "Ubuntu-3ubuntu0.6",
-      "protocol": "2.0",
-      "software": "OpenSSH_8.9p1"
-    },
-    "client": {
-      "comments": "IMHACKER",
-      "protocol": "2.0",
-      "software": "OpenSSH_8.9p1"
-    }
-  }
-}
-```
-
-Example for blocking all SSH connections:
-
-```yaml
-- name: Block SSH
-  action: block
-  expr: ssh != nil
-```
-
-## TLS
-
-```json
-{
-  "tls": {
-    "req": {
-      "alpn": ["h2", "http/1.1"],
-      "ciphers": [
-        4866, 4867, 4865, 49196, 49200, 159, 52393, 52392, 52394, 49195, 49199,
-        158, 49188, 49192, 107, 49187, 49191, 103, 49162, 49172, 57, 49161,
-        49171, 51, 157, 156, 61, 60, 53, 47, 255
-      ],
-      "compression": "AA==",
-      "random": "UqfPi+EmtMgusILrKcELvVWwpOdPSM/My09nPXl84dg=",
-      "session": "jCTrpAzHpwrfuYdYx4FEjZwbcQxCuZ52HGIoOcbw1vA=",
-      "sni": "ipinfo.io",
-      "supported_versions": [772, 771],
-      "version": 771,
-      "ech": true
-    },
-    "resp": {
-      "cipher": 4866,
-      "compression": 0,
-      "random": "R/Cy1m9pktuBMZQIHahD8Y83UWPRf8j8luwNQep9yJI=",
-      "session": "jCTrpAzHpwrfuYdYx4FEjZwbcQxCuZ52HGIoOcbw1vA=",
-      "supported_versions": 772,
-      "version": 771
-    }
-  }
-}
-```
-
-Example for blocking TLS connections to `ipinfo.io`:
-
-```yaml
-- name: Block ipinfo.io TLS
-  action: block
-  expr: tls != nil && tls.req != nil && tls.req.sni == "ipinfo.io"
-```
-
-## QUIC
-
-QUIC analyzer produces the same result format as TLS analyzer, but currently only supports "req" direction (client
-hello), not "resp" (server hello).
-
-```json
-{
-  "quic": {
-    "req": {
-      "alpn": ["h3"],
-      "ciphers": [4865, 4866, 4867],
-      "compression": "AA==",
-      "ech": true,
-      "random": "FUYLceFReLJl9dRQ0HAus7fi2ZGuKIAApF4keeUqg00=",
-      "session": "",
-      "sni": "quic.rocks",
-      "supported_versions": [772],
-      "version": 771
-    }
-  }
-}
-```
-
-Example for blocking QUIC connections to `quic.rocks`:
-
-```yaml
-- name: Block quic.rocks QUIC
-  action: block
-  expr: quic != nil && quic.req != nil && quic.req.sni == "quic.rocks"
-```
-
-## Trojan (proxy protocol)
-
-```json
-{
-  "trojan": {
-    "seq": [680, 4514, 293],
-    "yes": true
-  }
-}
-```
-
-Example for blocking Trojan connections:
-
-```yaml
-- name: Block Trojan
-  action: block
-  expr: trojan != nil && trojan.yes
-```
-
-## SOCKS
-
-SOCKS4:
-
-```json
-{
-  "socks": {
-    "version": 4,
-    "req": {
-      "cmd": 1,
-      "addr_type": 1, // same as socks5
-      "addr": "1.1.1.1",
-      // for socks4a
-      // "addr_type": 3,
-      // "addr": "google.com",
-      "port": 443,
-      "auth": {
-        "user_id": "user"
-      }
-    },
-    "resp": {
-      "rep": 90, // 0x5A(90) granted
-      "addr_type": 1,
-      "addr": "1.1.1.1",
-      "port": 443
-    }
-  }
-}
-```
-
-SOCKS5 without auth:
-
-```json
-{
-  "socks": {
-    "version": 5,
-    "req": {
-      "cmd": 1, // 0x01: connect, 0x02: bind, 0x03: udp
-      "addr_type": 3, // 0x01: ipv4, 0x03: domain, 0x04: ipv6
-      "addr": "google.com",
-      "port": 80,
-      "auth": {
-        "method": 0 // 0x00: no auth, 0x02: username/password
-      }
-    },
-    "resp": {
-      "rep": 0, // 0x00: success
-      "addr_type": 1, // 0x01: ipv4, 0x03: domain, 0x04: ipv6
-      "addr": "198.18.1.31",
-      "port": 80,
-      "auth": {
-        "method": 0 // 0x00: no auth, 0x02: username/password
-      }
-    }
-  }
-}
-```
-
-SOCKS5 with auth:
-
-```json
-{
-  "socks": {
-    "version": 5,
-    "req": {
-      "cmd": 1, // 0x01: connect, 0x02: bind, 0x03: udp
-      "addr_type": 3, // 0x01: ipv4, 0x03: domain, 0x04: ipv6
-      "addr": "google.com",
-      "port": 80,
-      "auth": {
-        "method": 2, // 0x00: no auth, 0x02: username/password
-        "username": "user",
-        "password": "pass"
-      }
-    },
-    "resp": {
-      "rep": 0, // 0x00: success
-      "addr_type": 1, // 0x01: ipv4, 0x03: domain, 0x04: ipv6
-      "addr": "198.18.1.31",
-      "port": 80,
-      "auth": {
-        "method": 2, // 0x00: no auth, 0x02: username/password
-        "status": 0 // 0x00: success, 0x01: failure
-      }
-    }
-  }
-}
-```
-
-Example for blocking connections to `google.com:80` and user `foobar`:
-
-```yaml
-- name: Block SOCKS google.com:80
-  action: block
-  expr: string(socks?.req?.addr) endsWith "google.com" && socks?.req?.port == 80
-
-- name: Block SOCKS user foobar
-  action: block
-  expr: socks?.req?.auth?.method == 2 && socks?.req?.auth?.username == "foobar"
-```
-
-## WireGuard
-
-```json
-{
-  "wireguard": {
-    "message_type": 1, // 0x1: handshake_initiation, 0x2: handshake_response, 0x3: packet_cookie_reply, 0x4: packet_data
-    "handshake_initiation": {
-      "sender_index": 0x12345678
-    },
-    "handshake_response": {
-      "sender_index": 0x12345678,
-      "receiver_index": 0x87654321,
-      "receiver_index_matched": true
-    },
-    "packet_data": {
-      "receiver_index": 0x12345678,
-      "receiver_index_matched": true
-    },
-    "packet_cookie_reply": {
-      "receiver_index": 0x12345678,
-      "receiver_index_matched": true
-    }
-  }
-}
-```
-
-Example for blocking WireGuard traffic:
-
-```yaml
-# false positive: high
-- name: Block all WireGuard-like traffic
-  action: block
-  expr: wireguard != nil
-
-# false positive: medium
-- name: Block WireGuard by handshake_initiation
-  action: drop
-  expr: wireguard?.handshake_initiation != nil
-
-# false positive: low
-- name: Block WireGuard by handshake_response
-  action: drop
-  expr: wireguard?.handshake_response?.receiver_index_matched == true
-
-# false positive: pretty low
-- name: Block WireGuard by packet_data
-  action: block
-  expr: wireguard?.packet_data?.receiver_index_matched == true
-```

engine/engine.go

@@ -15,7 +15,7 @@ var _ Engine = (*engine)(nil)
 
 type engine struct {
 	logger  Logger
-	ioList  []io.PacketIO
+	io      io.PacketIO
 	workers []*worker
 }
 
@@ -34,6 +34,7 @@ func NewEngine(config Config) (Engine, error) {
 			Ruleset:                    config.Ruleset,
 			TCPMaxBufferedPagesTotal:   config.WorkerTCPMaxBufferedPagesTotal,
 			TCPMaxBufferedPagesPerConn: config.WorkerTCPMaxBufferedPagesPerConn,
+			TCPTimeout:                 config.WorkerTCPTimeout,
 			UDPMaxStreams:              config.WorkerUDPMaxStreams,
 		})
 		if err != nil {
@@ -42,7 +43,7 @@ func NewEngine(config Config) (Engine, error) {
 	}
 	return &engine{
 		logger:  config.Logger,
-		ioList:  config.IOs,
+		io:      config.IO,
 		workers: workers,
 	}, nil
 }
@@ -58,27 +59,24 @@ func (e *engine) UpdateRuleset(r ruleset.Ruleset) error {
 
 func (e *engine) Run(ctx context.Context) error {
 	ioCtx, ioCancel := context.WithCancel(ctx)
-	defer ioCancel() // Stop workers & IOs
+	defer ioCancel() // Stop workers & IO
 
 	// Start workers
 	for _, w := range e.workers {
 		go w.Run(ioCtx)
 	}
 
-	// Register callbacks
-	errChan := make(chan error, len(e.ioList))
-	for _, i := range e.ioList {
-		ioEntry := i // Make sure dispatch() uses the correct ioEntry
-		err := ioEntry.Register(ioCtx, func(p io.Packet, err error) bool {
-			if err != nil {
-				errChan <- err
-				return false
-			}
-			return e.dispatch(ioEntry, p)
-		})
+	// Register IO callback
+	errChan := make(chan error, 1)
+	err := e.io.Register(ioCtx, func(p io.Packet, err error) bool {
 		if err != nil {
-			return err
+			errChan <- err
+			return false
 		}
+		return e.dispatch(p)
+	})
+	if err != nil {
+		return err
 	}
 
 	// Block until IO errors or context is cancelled
@@ -91,8 +89,7 @@ func (e *engine) Run(ctx context.Context) error {
 }
 
 // dispatch dispatches a packet to a worker.
-// This must be safe for concurrent use, as it may be called from multiple IOs.
-func (e *engine) dispatch(ioEntry io.PacketIO, p io.Packet) bool {
+func (e *engine) dispatch(p io.Packet) bool {
 	data := p.Data()
 	ipVersion := data[0] >> 4
 	var layerType gopacket.LayerType
@@ -102,17 +99,19 @@ func (e *engine) dispatch(ioEntry io.PacketIO, p io.Packet) bool {
 		layerType = layers.LayerTypeIPv6
 	} else {
 		// Unsupported network layer
-		_ = ioEntry.SetVerdict(p, io.VerdictAcceptStream, nil)
+		_ = e.io.SetVerdict(p, io.VerdictAcceptStream, nil)
 		return true
 	}
+	// Convert to gopacket.Packet
+	packet := gopacket.NewPacket(data, layerType, gopacket.DecodeOptions{Lazy: true, NoCopy: true})
+	packet.Metadata().Timestamp = p.Timestamp()
 	// Load balance by stream ID
 	index := p.StreamID() % uint32(len(e.workers))
-	packet := gopacket.NewPacket(data, layerType, gopacket.DecodeOptions{Lazy: true, NoCopy: true})
 	e.workers[index].Feed(&workerPacket{
 		StreamID: p.StreamID(),
 		Packet:   packet,
 		SetVerdict: func(v io.Verdict, b []byte) error {
-			return ioEntry.SetVerdict(p, v, b)
+			return e.io.SetVerdict(p, v, b)
 		},
 	})
 	return true

engine/interface.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"context"
+	"time"
 
 	"github.com/apernet/OpenGFW/io"
 	"github.com/apernet/OpenGFW/ruleset"
@@ -18,13 +19,14 @@ type Engine interface {
 // Config is the configuration for the engine.
 type Config struct {
 	Logger  Logger
-	IOs     []io.PacketIO
+	IO      io.PacketIO
 	Ruleset ruleset.Ruleset
 
 	Workers                          int // Number of workers. Zero or negative means auto (number of CPU cores).
 	WorkerQueueSize                  int
 	WorkerTCPMaxBufferedPagesTotal   int
 	WorkerTCPMaxBufferedPagesPerConn int
+	WorkerTCPTimeout                 time.Duration
 	WorkerUDPMaxStreams              int
 }
 
@@ -36,6 +38,7 @@ type Logger interface {
 	TCPStreamNew(workerID int, info ruleset.StreamInfo)
 	TCPStreamPropUpdate(info ruleset.StreamInfo, close bool)
 	TCPStreamAction(info ruleset.StreamInfo, action ruleset.Action, noMatch bool)
+	TCPFlush(workerID, flushed, closed int)
 
 	UDPStreamNew(workerID int, info ruleset.StreamInfo)
 	UDPStreamPropUpdate(info ruleset.StreamInfo, close bool)

engine/worker.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"context"
+	"time"
 
 	"github.com/apernet/OpenGFW/io"
 	"github.com/apernet/OpenGFW/ruleset"
@@ -14,9 +15,12 @@ import (
 
 const (
 	defaultChanSize                         = 64
-	defaultTCPMaxBufferedPagesTotal         = 4096
-	defaultTCPMaxBufferedPagesPerConnection = 64
+	defaultTCPMaxBufferedPagesTotal         = 65536
+	defaultTCPMaxBufferedPagesPerConnection = 16
+	defaultTCPTimeout                       = 10 * time.Minute
 	defaultUDPMaxStreams                    = 4096
+
+	tcpFlushInterval = 1 * time.Minute
 )
 
 type workerPacket struct {
@@ -33,6 +37,7 @@ type worker struct {
 	tcpStreamFactory *tcpStreamFactory
 	tcpStreamPool    *reassembly.StreamPool
 	tcpAssembler     *reassembly.Assembler
+	tcpTimeout       time.Duration
 
 	udpStreamFactory *udpStreamFactory
 	udpStreamManager *udpStreamManager
@@ -47,6 +52,7 @@ type workerConfig struct {
 	Ruleset                    ruleset.Ruleset
 	TCPMaxBufferedPagesTotal   int
 	TCPMaxBufferedPagesPerConn int
+	TCPTimeout                 time.Duration
 	UDPMaxStreams              int
 }
 
@@ -60,6 +66,9 @@ func (c *workerConfig) fillDefaults() {
 	if c.TCPMaxBufferedPagesPerConn <= 0 {
 		c.TCPMaxBufferedPagesPerConn = defaultTCPMaxBufferedPagesPerConnection
 	}
+	if c.TCPTimeout <= 0 {
+		c.TCPTimeout = defaultTCPTimeout
+	}
 	if c.UDPMaxStreams <= 0 {
 		c.UDPMaxStreams = defaultUDPMaxStreams
 	}
@@ -98,6 +107,7 @@ func newWorker(config workerConfig) (*worker, error) {
 		tcpStreamFactory:   tcpSF,
 		tcpStreamPool:      tcpStreamPool,
 		tcpAssembler:       tcpAssembler,
+		tcpTimeout:         config.TCPTimeout,
 		udpStreamFactory:   udpSF,
 		udpStreamManager:   udpSM,
 		modSerializeBuffer: gopacket.NewSerializeBuffer(),
@@ -111,6 +121,10 @@ func (w *worker) Feed(p *workerPacket) {
 func (w *worker) Run(ctx context.Context) {
 	w.logger.WorkerStart(w.id)
 	defer w.logger.WorkerStop(w.id)
+
+	tcpFlushTicker := time.NewTicker(tcpFlushInterval)
+	defer tcpFlushTicker.Stop()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -122,6 +136,8 @@ func (w *worker) Run(ctx context.Context) {
 			}
 			v, b := w.handle(wPkt.StreamID, wPkt.Packet)
 			_ = wPkt.SetVerdict(v, b)
+		case <-tcpFlushTicker.C:
+			w.flushTCP(w.tcpTimeout)
 		}
 	}
 }
@@ -176,6 +192,11 @@ func (w *worker) handleTCP(ipFlow gopacket.Flow, pMeta *gopacket.PacketMetadata,
 	return io.Verdict(ctx.Verdict)
 }
 
+func (w *worker) flushTCP(timeout time.Duration) {
+	flushed, closed := w.tcpAssembler.FlushCloseOlderThan(time.Now().Add(-timeout))
+	w.logger.TCPFlush(w.id, flushed, closed)
+}
+
 func (w *worker) handleUDP(streamID uint32, ipFlow gopacket.Flow, udp *layers.UDP) (io.Verdict, []byte) {
 	ctx := &udpContext{
 		Verdict: udpVerdictAccept,

go.mod

@@ -5,7 +5,7 @@ go 1.21
 require (
 	github.com/bwmarrin/snowflake v0.3.0
 	github.com/coreos/go-iptables v0.7.0
-	github.com/expr-lang/expr v1.15.7
+	github.com/expr-lang/expr v1.16.3
 	github.com/florianl/go-nfqueue v1.3.2-0.20231218173729-f2bdeb033acf
 	github.com/google/gopacket v1.1.20-0.20220810144506-32ee38206866
 	github.com/hashicorp/golang-lru/v2 v2.0.7

go.sum

@@ -7,8 +7,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/expr-lang/expr v1.15.7 h1:BK0JcWUkoW6nrbLBo6xCKhz4BvH5DSOOu1Gx5lucyZo=
-github.com/expr-lang/expr v1.15.7/go.mod h1:uCkhfG+x7fcZ5A5sXHKuQ07jGZRl6J0FCAaf2k4PtVQ=
+github.com/expr-lang/expr v1.16.3 h1:NLldf786GffptcXNxxJx5dQ+FzeWDKChBDqOOwyK8to=
+github.com/expr-lang/expr v1.16.3/go.mod h1:uCkhfG+x7fcZ5A5sXHKuQ07jGZRl6J0FCAaf2k4PtVQ=
 github.com/florianl/go-nfqueue v1.3.2-0.20231218173729-f2bdeb033acf h1:NqGS3vTHzVENbIfd87cXZwdpO6MB2R1PjHMJLi4Z3ow=
 github.com/florianl/go-nfqueue v1.3.2-0.20231218173729-f2bdeb033acf/go.mod h1:eSnAor2YCfMCVYrVNEhkLGN/r1L+J4uDjc0EUy0tfq4=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=

io/interface.go

@@ -2,6 +2,8 @@ package io
 
 import (
 	"context"
+	"net"
+	"time"
 )
 
 type Verdict int
@@ -23,13 +25,14 @@ const (
 type Packet interface {
 	// StreamID is the ID of the stream the packet belongs to.
 	StreamID() uint32
+	// Timestamp is the time the packet was received.
+	Timestamp() time.Time
 	// Data is the raw packet data, starting with the IP header.
 	Data() []byte
 }
 
 // PacketCallback is called for each packet received.
 // Return false to "unregister" and stop receiving packets.
-// It must be safe for concurrent use.
 type PacketCallback func(Packet, error) bool
 
 type PacketIO interface {
@@ -39,6 +42,10 @@ type PacketIO interface {
 	Register(context.Context, PacketCallback) error
 	// SetVerdict sets the verdict for a packet.
 	SetVerdict(Packet, Verdict, []byte) error
+	// ProtectedDialContext is like net.DialContext, but the connection is "protected"
+	// in the sense that the packets sent/received through the connection must bypass
+	// the packet IO and not be processed by the callback.
+	ProtectedDialContext(ctx context.Context, network, address string) (net.Conn, error)
 	// Close closes the packet IO.
 	Close() error
 }

io/nfqueue.go

@@ -5,9 +5,12 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"net"
 	"os/exec"
 	"strconv"
 	"strings"
+	"syscall"
+	"time"
 
 	"github.com/coreos/go-iptables/iptables"
 	"github.com/florianl/go-nfqueue"
@@ -50,6 +53,7 @@ func generateNftRules(local, rst bool) (*nftTableSpec, error) {
 	}
 	for i := range table.Chains {
 		c := &table.Chains[i]
+		c.Rules = append(c.Rules, "meta mark $ACCEPT_CTMARK ct mark set $ACCEPT_CTMARK") // Bypass protected connections
 		c.Rules = append(c.Rules, "ct mark $ACCEPT_CTMARK counter accept")
 		if rst {
 			c.Rules = append(c.Rules, "ip protocol tcp ct mark $DROP_CTMARK counter reject with tcp reset")
@@ -72,6 +76,8 @@ func generateIptRules(local, rst bool) ([]iptRule, error) {
 	}
 	rules := make([]iptRule, 0, 4*len(chains))
 	for _, chain := range chains {
+		// Bypass protected connections
+		rules = append(rules, iptRule{"filter", chain, []string{"-m", "mark", "--mark", strconv.Itoa(nfqueueConnMarkAccept), "-j", "CONNMARK", "--set-mark", strconv.Itoa(nfqueueConnMarkAccept)}})
 		rules = append(rules, iptRule{"filter", chain, []string{"-m", "connmark", "--mark", strconv.Itoa(nfqueueConnMarkAccept), "-j", "ACCEPT"}})
 		if rst {
 			rules = append(rules, iptRule{"filter", chain, []string{"-p", "tcp", "-m", "connmark", "--mark", strconv.Itoa(nfqueueConnMarkDrop), "-j", "REJECT", "--reject-with", "tcp-reset"}})
@@ -96,6 +102,8 @@ type nfqueuePacketIO struct {
 	// iptables not nil = use iptables instead of nftables
 	ipt4 *iptables.IPTables
 	ipt6 *iptables.IPTables
+
+	protectedDialer *net.Dialer
 }
 
 type NFQueuePacketIOConfig struct {
@@ -153,6 +161,18 @@ func NewNFQueuePacketIO(config NFQueuePacketIOConfig) (PacketIO, error) {
 		rst:   config.RST,
 		ipt4:  ipt4,
 		ipt6:  ipt6,
+		protectedDialer: &net.Dialer{
+			Control: func(network, address string, c syscall.RawConn) error {
+				var err error
+				cErr := c.Control(func(fd uintptr) {
+					err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_MARK, nfqueueConnMarkAccept)
+				})
+				if cErr != nil {
+					return cErr
+				}
+				return err
+			},
+		},
 	}, nil
 }
 
@@ -170,6 +190,12 @@ func (n *nfqueuePacketIO) Register(ctx context.Context, cb PacketCallback) error
 				streamID: ctIDFromCtBytes(*a.Ct),
 				data:     *a.Payload,
 			}
+			// Use timestamp from attribute if available, otherwise use current time as fallback
+			if a.Timestamp != nil {
+				p.timestamp = *a.Timestamp
+			} else {
+				p.timestamp = time.Now()
+			}
 			return okBoolToInt(cb(p, nil))
 		},
 		func(e error) int {
@@ -239,6 +265,10 @@ func (n *nfqueuePacketIO) SetVerdict(p Packet, v Verdict, newPacket []byte) erro
 	}
 }
 
+func (n *nfqueuePacketIO) ProtectedDialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	return n.protectedDialer.DialContext(ctx, network, address)
+}
+
 func (n *nfqueuePacketIO) Close() error {
 	if n.rSet {
 		if n.ipt4 != nil {
@@ -289,15 +319,20 @@ func (n *nfqueuePacketIO) setupIpt(local, rst, remove bool) error {
 var _ Packet = (*nfqueuePacket)(nil)
 
 type nfqueuePacket struct {
-	id       uint32
-	streamID uint32
-	data     []byte
+	id        uint32
+	streamID  uint32
+	timestamp time.Time
+	data      []byte
 }
 
 func (p *nfqueuePacket) StreamID() uint32 {
 	return p.streamID
 }
 
+func (p *nfqueuePacket) Timestamp() time.Time {
+	return p.timestamp
+}
+
 func (p *nfqueuePacket) Data() []byte {
 	return p.data
 }

ruleset/builtins/geo/geo_matcher.go

@@ -14,14 +14,12 @@ type GeoMatcher struct {
 	ipMatcherLock   sync.Mutex
 }
 
-func NewGeoMatcher(geoSiteFilename, geoIpFilename string) (*GeoMatcher, error) {
-	geoLoader := NewDefaultGeoLoader(geoSiteFilename, geoIpFilename)
-
+func NewGeoMatcher(geoSiteFilename, geoIpFilename string) *GeoMatcher {
 	return &GeoMatcher{
-		geoLoader:      geoLoader,
+		geoLoader:      NewDefaultGeoLoader(geoSiteFilename, geoIpFilename),
 		geoSiteMatcher: make(map[string]hostMatcher),
 		geoIpMatcher:   make(map[string]hostMatcher),
-	}, nil
+	}
 }
 
 func (g *GeoMatcher) MatchGeoIp(ip, condition string) bool {

ruleset/expr.go

@@ -1,11 +1,15 @@
 package ruleset
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"os"
 	"reflect"
 	"strings"
+	"time"
+
+	"github.com/expr-lang/expr/builtin"
 
 	"github.com/expr-lang/expr"
 	"github.com/expr-lang/expr/ast"
@@ -16,7 +20,6 @@ import (
 	"github.com/apernet/OpenGFW/analyzer"
 	"github.com/apernet/OpenGFW/modifier"
 	"github.com/apernet/OpenGFW/ruleset/builtins"
-	"github.com/apernet/OpenGFW/ruleset/builtins/geo"
 )
 
 // ExprRule is the external representation of an expression rule.
@@ -55,10 +58,9 @@ type compiledExprRule struct {
 var _ Ruleset = (*exprRuleset)(nil)
 
 type exprRuleset struct {
-	Rules      []compiledExprRule
-	Ans        []analyzer.Analyzer
-	Logger     Logger
-	GeoMatcher *geo.GeoMatcher
+	Rules  []compiledExprRule
+	Ans    []analyzer.Analyzer
+	Logger Logger
 }
 
 func (r *exprRuleset) Analyzers(info StreamInfo) []analyzer.Analyzer {
@@ -100,10 +102,7 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 	fullAnMap := analyzersToMap(ans)
 	fullModMap := modifiersToMap(mods)
 	depAnMap := make(map[string]analyzer.Analyzer)
-	geoMatcher, err := geo.NewGeoMatcher(config.GeoSiteFilename, config.GeoIpFilename)
-	if err != nil {
-		return nil, err
-	}
+	funcMap := buildFunctionMap(config)
 	// Compile all rules and build a map of analyzers that are used by the rules.
 	for _, rule := range rules {
 		if rule.Action == "" && !rule.Log {
@@ -118,13 +117,19 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 			action = &a
 		}
 		visitor := &idVisitor{Variables: make(map[string]bool), Identifiers: make(map[string]bool)}
-		patcher := &idPatcher{}
+		patcher := &idPatcher{FuncMap: funcMap}
 		program, err := expr.Compile(rule.Expr,
 			func(c *conf.Config) {
 				c.Strict = false
 				c.Expect = reflect.Bool
 				c.Visitors = append(c.Visitors, visitor, patcher)
-				registerBuiltinFunctions(c.Functions, geoMatcher)
+				for name, f := range funcMap {
+					c.Functions[name] = &builtin.Function{
+						Name:  name,
+						Func:  f.Func,
+						Types: f.Types,
+					}
+				}
 			},
 		)
 		if err != nil {
@@ -138,24 +143,15 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 			if isBuiltInAnalyzer(name) || visitor.Variables[name] {
 				continue
 			}
-			// Check if it's one of the built-in functions, and if so,
-			// skip it as an analyzer & do initialization if necessary.
-			switch name {
-			case "geoip":
-				if err := geoMatcher.LoadGeoIP(); err != nil {
-					return nil, fmt.Errorf("rule %q failed to load geoip: %w", rule.Name, err)
-				}
-			case "geosite":
-				if err := geoMatcher.LoadGeoSite(); err != nil {
-					return nil, fmt.Errorf("rule %q failed to load geosite: %w", rule.Name, err)
-				}
-			case "cidr":
-				// No initialization needed for CIDR.
-			default:
-				a, ok := fullAnMap[name]
-				if !ok {
-					return nil, fmt.Errorf("rule %q uses unknown analyzer %q", rule.Name, name)
+			if f, ok := funcMap[name]; ok {
+				// Built-in function, initialize if necessary
+				if f.InitFunc != nil {
+					if err := f.InitFunc(); err != nil {
+						return nil, fmt.Errorf("rule %q failed to initialize function %q: %w", rule.Name, name, err)
+					}
 				}
+			} else if a, ok := fullAnMap[name]; ok {
+				// Analyzer, add to dependency map
 				depAnMap[name] = a
 			}
 		}
@@ -184,37 +180,12 @@ func CompileExprRules(rules []ExprRule, ans []analyzer.Analyzer, mods []modifier
 		depAns = append(depAns, a)
 	}
 	return &exprRuleset{
-		Rules:      compiledRules,
-		Ans:        depAns,
-		Logger:     config.Logger,
-		GeoMatcher: geoMatcher,
+		Rules:  compiledRules,
+		Ans:    depAns,
+		Logger: config.Logger,
 	}, nil
 }
 
-func registerBuiltinFunctions(funcMap map[string]*ast.Function, geoMatcher *geo.GeoMatcher) {
-	funcMap["geoip"] = &ast.Function{
-		Name: "geoip",
-		Func: func(params ...any) (any, error) {
-			return geoMatcher.MatchGeoIp(params[0].(string), params[1].(string)), nil
-		},
-		Types: []reflect.Type{reflect.TypeOf(geoMatcher.MatchGeoIp)},
-	}
-	funcMap["geosite"] = &ast.Function{
-		Name: "geosite",
-		Func: func(params ...any) (any, error) {
-			return geoMatcher.MatchGeoSite(params[0].(string), params[1].(string)), nil
-		},
-		Types: []reflect.Type{reflect.TypeOf(geoMatcher.MatchGeoSite)},
-	}
-	funcMap["cidr"] = &ast.Function{
-		Name: "cidr",
-		Func: func(params ...any) (any, error) {
-			return builtins.MatchCIDR(params[0].(string), params[1].(*net.IPNet)), nil
-		},
-		Types: []reflect.Type{reflect.TypeOf((func(string, string) bool)(nil)), reflect.TypeOf(builtins.MatchCIDR)},
-	}
-}
-
 func streamInfoToExprEnv(info StreamInfo) map[string]interface{} {
 	m := map[string]interface{}{
 		"id":    info.ID,
@@ -299,29 +270,108 @@ func (v *idVisitor) Visit(node *ast.Node) {
 // idPatcher patches the AST during expr compilation, replacing certain values with
 // their internal representations for better runtime performance.
 type idPatcher struct {
-	Err error
+	FuncMap map[string]*Function
+	Err     error
 }
 
 func (p *idPatcher) Visit(node *ast.Node) {
 	switch (*node).(type) {
 	case *ast.CallNode:
 		callNode := (*node).(*ast.CallNode)
-		if callNode.Func == nil {
+		if callNode.Callee == nil {
 			// Ignore invalid call nodes
 			return
 		}
-		switch callNode.Func.Name {
-		case "cidr":
-			cidrStringNode, ok := callNode.Arguments[1].(*ast.StringNode)
-			if !ok {
-				return
+		if f, ok := p.FuncMap[callNode.Callee.String()]; ok {
+			if f.PatchFunc != nil {
+				if err := f.PatchFunc(&callNode.Arguments); err != nil {
+					p.Err = err
+					return
+				}
 			}
-			cidr, err := builtins.CompileCIDR(cidrStringNode.Value)
-			if err != nil {
-				p.Err = err
-				return
-			}
-			callNode.Arguments[1] = &ast.ConstantNode{Value: cidr}
 		}
 	}
 }
+
+type Function struct {
+	InitFunc  func() error
+	PatchFunc func(args *[]ast.Node) error
+	Func      func(params ...any) (any, error)
+	Types     []reflect.Type
+}
+
+func buildFunctionMap(config *BuiltinConfig) map[string]*Function {
+	return map[string]*Function{
+		"geoip": {
+			InitFunc:  config.GeoMatcher.LoadGeoIP,
+			PatchFunc: nil,
+			Func: func(params ...any) (any, error) {
+				return config.GeoMatcher.MatchGeoIp(params[0].(string), params[1].(string)), nil
+			},
+			Types: []reflect.Type{reflect.TypeOf(config.GeoMatcher.MatchGeoIp)},
+		},
+		"geosite": {
+			InitFunc:  config.GeoMatcher.LoadGeoSite,
+			PatchFunc: nil,
+			Func: func(params ...any) (any, error) {
+				return config.GeoMatcher.MatchGeoSite(params[0].(string), params[1].(string)), nil
+			},
+			Types: []reflect.Type{reflect.TypeOf(config.GeoMatcher.MatchGeoSite)},
+		},
+		"cidr": {
+			InitFunc: nil,
+			PatchFunc: func(args *[]ast.Node) error {
+				cidrStringNode, ok := (*args)[1].(*ast.StringNode)
+				if !ok {
+					return fmt.Errorf("cidr: invalid argument type")
+				}
+				cidr, err := builtins.CompileCIDR(cidrStringNode.Value)
+				if err != nil {
+					return err
+				}
+				(*args)[1] = &ast.ConstantNode{Value: cidr}
+				return nil
+			},
+			Func: func(params ...any) (any, error) {
+				return builtins.MatchCIDR(params[0].(string), params[1].(*net.IPNet)), nil
+			},
+			Types: []reflect.Type{reflect.TypeOf(builtins.MatchCIDR)},
+		},
+		"lookup": {
+			InitFunc: nil,
+			PatchFunc: func(args *[]ast.Node) error {
+				var serverStr *ast.StringNode
+				if len(*args) > 1 {
+					// Has the optional server argument
+					var ok bool
+					serverStr, ok = (*args)[1].(*ast.StringNode)
+					if !ok {
+						return fmt.Errorf("lookup: invalid argument type")
+					}
+				}
+				r := &net.Resolver{
+					Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
+						if serverStr != nil {
+							address = serverStr.Value
+						}
+						return config.ProtectedDialContext(ctx, network, address)
+					},
+				}
+				if len(*args) > 1 {
+					(*args)[1] = &ast.ConstantNode{Value: r}
+				} else {
+					*args = append(*args, &ast.ConstantNode{Value: r})
+				}
+				return nil
+			},
+			Func: func(params ...any) (any, error) {
+				ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+				defer cancel()
+				return params[1].(*net.Resolver).LookupHost(ctx, params[0].(string))
+			},
+			Types: []reflect.Type{
+				reflect.TypeOf((func(string, *net.Resolver) []string)(nil)),
+			},
+		},
+	}
+}

ruleset/interface.go

@@ -1,11 +1,13 @@
 package ruleset
 
 import (
+	"context"
 	"net"
 	"strconv"
 
 	"github.com/apernet/OpenGFW/analyzer"
 	"github.com/apernet/OpenGFW/modifier"
+	"github.com/apernet/OpenGFW/ruleset/builtins/geo"
 )
 
 type Action int
@@ -100,7 +102,7 @@ type Logger interface {
 }
 
 type BuiltinConfig struct {
-	Logger          Logger
-	GeoSiteFilename string
-	GeoIpFilename   string
+	Logger               Logger
+	GeoMatcher           *geo.GeoMatcher
+	ProtectedDialContext func(ctx context.Context, network, address string) (net.Conn, error)
 }