Block BitTorrent on the server

- awg-ctrl: optional nft table awg_torrent (GET/PUT /api/torrent, off by
  default). A connection is judged by its first packets, like Xray's
  sniffing: DHT, UDP trackers, uTP SYN, and with @ih support the plain
  TCP handshake and HTTP announce. A match marks the connection in
  conntrack and drops it whole; without @ih only the UDP part is used
- devices that tried are kept in a 10-minute set and flagged as
  "torrent" in user and device stats; with shaping on they sit in the
  background class while the flag lasts
- metrics: awg_torrent_block_enabled/active, awg_torrent_blocked_total,
  awg_torrent_devices and per-peer awg_peer_torrent
- panel: "Торренты" section in server settings, "торренты" chip on users
  and devices; Save sends only what changed
This commit is contained in:
2026-10-17 23:35:30 +03:00
parent cf6af5ce0a
commit c73da602d6
16 changed files with 575 additions and 55 deletions
+5 -2
View File
@@ -31,6 +31,7 @@ type metricsPeer struct {
Rx float64 `json:"rx"`
Tx float64 `json:"tx"`
Priority string `json:"priority"`
Torrent bool `json:"torrent"`
}
type metricsHost struct {
@@ -65,6 +66,7 @@ type metricsBody struct {
TxBytes float64 `json:"tx_bytes"`
Link linkStats `json:"link"`
Shaping ShapingStatus `json:"shaping"`
Torrent TorrentStatus `json:"torrent"`
Host *metricsHost `json:"host"`
List []metricsPeer `json:"list"`
}
@@ -139,7 +141,7 @@ func (s *Server) collectMetrics() (*metricsBody, error) {
return nil, err
}
ps := stats[pub]
b.List = append(b.List, metricsPeer{"user", n, ip, ps.Online, orZero(ps.Rx), orZero(ps.Tx), s.shaping.PriorityOf(ip)})
b.List = append(b.List, metricsPeer{"user", n, ip, ps.Online, orZero(ps.Rx), orZero(ps.Tx), s.shaping.PriorityOf(ip), s.torrent.Seen(ip)})
}
rows.Close()
devs, err := s.peersAll()
@@ -148,7 +150,7 @@ func (s *Server) collectMetrics() (*metricsBody, error) {
}
for _, p := range devs {
ps := stats[p.PubKey]
b.List = append(b.List, metricsPeer{"device", "", p.IP, ps.Online, orZero(ps.Rx), orZero(ps.Tx), s.shaping.PriorityOf(p.IP)})
b.List = append(b.List, metricsPeer{"device", "", p.IP, ps.Online, orZero(ps.Rx), orZero(ps.Tx), s.shaping.PriorityOf(p.IP), s.torrent.Seen(p.IP)})
}
b.Peers.Total = st.Peers
for _, ps := range stats {
@@ -163,6 +165,7 @@ func (s *Server) collectMetrics() (*metricsBody, error) {
s.linkCounter("rx_dropped"), s.linkCounter("tx_dropped"), s.linkCounter("rx_errors"), s.linkCounter("tx_errors"),
}
b.Shaping = s.shaping.Status()
b.Torrent = s.torrent.Status()
b.Host = readHost()
return b, nil
}
+18 -2
View File
@@ -87,7 +87,8 @@ func (s *Server) routes() []route {
}
ps := stats[pub]
list = append(list, ob{{"name", n}, {"ip", ip}, {"key_gen", kg}, {"online", ps.Online},
{"lastHandshake", ps.LastHandshake}, {"rx", ps.Rx}, {"tx", ps.Tx}, {"priority", s.shaping.PriorityOf(ip)}})
{"lastHandshake", ps.LastHandshake}, {"rx", ps.Rx}, {"tx", ps.Tx}, {"priority", s.shaping.PriorityOf(ip)},
{"torrent", s.torrent.Seen(ip)}})
}
rows.Close()
return ok(ob{{"users", list}})
@@ -147,7 +148,7 @@ func (s *Server) routes() []route {
ps := stats[p.PubKey]
list = append(list, ob{{"pub_key", p.PubKey}, {"ip", p.IP}, {"psk_key", p.PskKey}, {"owner", p.Owner},
{"online", ps.Online}, {"lastHandshake", ps.LastHandshake}, {"rx", ps.Rx}, {"tx", ps.Tx},
{"priority", s.shaping.PriorityOf(p.IP)}})
{"priority", s.shaping.PriorityOf(p.IP)}, {"torrent", s.torrent.Seen(p.IP)}})
}
return ok(ob{{"peers", list}})
}),
@@ -357,6 +358,21 @@ func (s *Server) routes() []route {
return ok(s.shaping.Status())
}),
// Блокировка BitTorrent. Тоже под /api/ и через PUT — чтобы проходила через awg-agent на ноде.
R("GET", "/api/torrent", false, a, func(c *rctx) (result, error) { return ok(s.torrent.Status()) }),
R("PUT", "/api/torrent", false, a, func(c *rctx) (result, error) {
enabled, isBool := c.obj().Get2("enabled").(bool)
if !isBool {
return result{}, httpErr(400, "enabled должен быть true/false")
}
// nft не нашёлся или ядро отказало: оператору нужна причина, а не 500.
if err := s.torrent.Update(TorrentSettings{Enabled: enabled}); err != nil {
return result{}, httpErr(500, err.Error())
}
return ok(s.torrent.Status())
}),
R("GET", "/api/metrics", false, a, func(c *rctx) (result, error) {
d, err := s.metricsCached()
if err != nil {
+3
View File
@@ -55,6 +55,7 @@ type Server struct {
c Config
shaping *Shaping
torrent *Torrent
metrics metricsCache
monitor monitorJob
cliPath string
@@ -155,6 +156,7 @@ func New() (*Server, error) {
}
s.log.Info("awg params loaded from conf", "Jc", params.Jc, "H1", params.H1, "gen", gen)
s.shaping = newShaping(s)
s.torrent = newTorrent(s)
return s, nil
}
@@ -182,5 +184,6 @@ func (s *Server) Run() error {
// ни один из путей awg-quick up внутри awg-ctrl тогда не срабатывает.
go s.shaping.Apply()
s.shaping.Start()
go s.torrent.Apply() // nft-таблицы после перезагрузки нет — ставим заново
return <-errc
}
+55 -26
View File
@@ -20,6 +20,7 @@ import (
// загрузка (к клиенту) — CAKE root на awg1, dual-dsthost, diffserv3
// отдача (от клиента) — ingress awg1 → ifb-awg, CAKE dual-srchost, besteffort
// тяжёлые при перегрузке — DSCP CS1 (bulk-tin CAKE) через отдельную nft-таблицу
// с меткой «торренты» — туда же, пока метка висит (см. torrent.go), независимо от перегрузки
//
// Ничего не пишет в awg1.conf (PostUp не трогаем): всё применяется из процесса и
// идемпотентно. Правила tc пропадают вместе с интерфейсом, поэтому Apply зовётся на
@@ -123,6 +124,7 @@ type Shaping struct {
histOrder []string
totals []tick
heavy *orderedSet
applied *orderedSet // что сейчас стоит в nft-наборе: тяжёлые ∪ с меткой торрента
congested bool
lastRateDn float64
lastRateUp float64
@@ -138,7 +140,7 @@ func newShaping(s *Server) *Shaping {
sampleS := envNum("SHAPE_SAMPLE_S", 10)
windowS := envNum("SHAPE_WINDOW_S", 600)
return &Shaping{
s: s, prev: map[string][2]float64{}, history: map[string][]tick{}, heavy: newOrderedSet(),
s: s, prev: map[string][2]float64{}, history: map[string][]tick{}, heavy: newOrderedSet(), applied: newOrderedSet(),
sampleS: sampleS, windowTicks: int(math.Max(3, math.Round(windowS/sampleS))),
}
}
@@ -266,13 +268,12 @@ func (h *Shaping) applyNow() error {
return err
}
}
h.mu.Lock()
heavy := h.heavy.list()
h.mu.Unlock()
heavy := h.desired()
if err := h.nftApply(heavy); err != nil {
return err
}
h.mu.Lock()
h.applied = setOf(heavy)
h.lastError = ""
h.mu.Unlock()
h.s.log.Info("shaping applied", "down", st.DownMbit, "up", st.UpMbit, "overhead", oh, "heavy", len(heavy))
@@ -287,6 +288,7 @@ func (h *Shaping) removeNow() {
shExec("nft", "delete", "table", "inet", nftTable)
h.mu.Lock()
h.heavy.clear()
h.applied.clear()
h.history, h.histOrder, h.prev = map[string][]tick{}, nil, map[string][2]float64{}
h.totals = nil
h.congested = false
@@ -319,6 +321,50 @@ func (h *Shaping) Update(st ShapingSettings) error {
return err
}
// desired — кого держать в фоновом классе: тяжёлые (порядок сэмплера) и, следом, устройства с
// меткой «торренты». Метку ставит и снимает nft (таймаут), поэтому сверяемся каждый тик.
func (h *Shaping) desired() []string {
h.mu.Lock()
out := setOf(h.heavy.list())
h.mu.Unlock()
for _, ip := range h.s.torrent.SeenList() {
out.add(ip)
}
return out.list()
}
func setOf(list []string) *orderedSet {
o := newOrderedSet()
for _, k := range list {
o.add(k)
}
return o
}
// syncNft — привести nft-набор к desired(); без изменений nft не трогаем.
func (h *Shaping) syncNft() {
want := h.desired()
h.mu.Lock()
same := len(want) == h.applied.size()
for _, ip := range want {
same = same && h.applied.has(ip)
}
h.mu.Unlock()
if same {
return
}
h.opMu.Lock()
err := h.nftApply(want)
h.opMu.Unlock()
if err != nil {
h.noteError(err)
return
}
h.mu.Lock()
h.applied = setOf(want)
h.mu.Unlock()
}
// ───────── сэмплер (RFC 6057) ─────────
func (h *Shaping) pushTick(arr []tick, t tick) []tick {
@@ -391,6 +437,8 @@ func (h *Shaping) tick() {
h.tickNo++
tickNo := h.tickNo
h.mu.Unlock()
// В конце любого тика, в том числе пока окно не набралось: метка торрента не ждёт 10 минут.
defer h.syncNft()
// Страховка: интерфейс могли поднять руками (awg-quick down/up) — qdisc тогда исчез.
if tickNo%checkEvery == 0 && !h.hasCake(h.s.iface) {
@@ -443,18 +491,9 @@ func (h *Shaping) tick() {
h.lastRateUp = totalUp * 8 / dt / 1e6
if !st.Auto {
changed := h.heavy.size() > 0 || h.congested
h.heavy.clear()
h.congested = false
h.mu.Unlock()
if changed {
h.opMu.Lock()
err := h.nftApply(nil)
h.opMu.Unlock()
if err != nil {
h.noteError(err)
}
}
return
}
if len(h.totals) < h.windowTicks { // пока окно не набралось, решений не принимаем
@@ -507,21 +546,11 @@ func (h *Shaping) tick() {
}
}
}
var apply []string
if changed || wasCongested != h.congested {
h.heavy = next
apply = next.list()
h.s.log.Info("shaping: класс тяжёлых обновлён", "congested", h.congested, "avgMbit", math.Round(avg), "heavy", apply)
h.s.log.Info("shaping: класс тяжёлых обновлён", "congested", h.congested, "avgMbit", math.Round(avg), "heavy", next.list())
}
h.mu.Unlock()
if changed || wasCongested != h.congested {
h.opMu.Lock()
err := h.nftApply(apply)
h.opMu.Unlock()
if err != nil {
h.noteError(err)
}
}
}
func (h *Shaping) Start() {
@@ -571,7 +600,7 @@ func (h *Shaping) qdiscStats(dev string) *QdiscStats {
func (h *Shaping) PriorityOf(ip string) string {
h.mu.Lock()
defer h.mu.Unlock()
if h.heavy.has(ip) {
if h.applied.has(ip) { // то, что реально стоит в nft: тяжёлые и с меткой торрента
return "background"
}
return "normal"
@@ -593,7 +622,7 @@ func (h *Shaping) Status() ShapingStatus {
devices++
}
}
heavy := h.heavy.list()
heavy := h.applied.list()
if heavy == nil {
heavy = []string{}
}
+294
View File
@@ -0,0 +1,294 @@
package ctrl
import (
"encoding/json"
"sort"
"strings"
"sync"
"time"
)
// Блокировка BitTorrent — то же, что sniffing + blackhole в Xray (3x-ui), только в ядре.
//
// Решение принимается по первым байтам и один раз на соединение: совпала сигнатура — соединение
// получает бит в ct mark и дальше режется целиком, в обе стороны. Смотрим только то, что
// клиенты шлют в интернет (iifname awg1), до NAT, поэтому источник — IP клиента в туннеле.
//
// Ловится: DHT (запросы и ответы), UDP-трекеры, начало uTP-соединения, а при поддержке @ih в
// nft/ядре — ещё открытое TCP-рукопожатие и HTTP-анонс. Не ловится шифрованное рукопожатие
// (MSE/PE) и HTTPS-трекеры — так же, как у Xray. Но без DHT и трекеров клиенту негде брать пиров.
//
// Как и шейпинг, ничего не пишет в awg1.conf: таблица ставится из процесса и идемпотентно.
// nft-таблица не привязана к интерфейсу (iifname — по имени), поэтому переживает его рестарт;
// после перезагрузки сервера её возвращает Apply на старте.
type TorrentSettings struct {
Enabled bool `json:"enabled"`
}
type TorrentPeer struct {
IP string `json:"ip"`
Name string `json:"name"` // имя пользователя; у sen://-устройства — пусто
}
type TorrentStatus struct {
Settings TorrentSettings `json:"settings"`
Active bool `json:"active"` // таблица реально стоит
TCP bool `json:"tcp"` // проверяется и TCP (нужен @ih: nftables ≥ 1.0, свежее ядро)
Hits float64 `json:"hits"` // заблокированных попыток с последнего применения
Seen []TorrentPeer `json:"seen"` // кто пытался за последние torrentSeenTTL
Error string `json:"error,omitempty"`
}
const (
torrentTable = "awg_torrent"
torrentMark = "0x00100000" // свой бит ct mark: чужие биты не трогаем
torrentSeenTTL = "10m"
torrentCacheS = 5 * time.Second // /api/users/stats опрашивается часто — nft не дёргаем на каждый
)
// Сигнатуры. @th,64 — начало данных UDP (заголовок 8 байт); @ih,0 — начало данных TCP с учётом опций.
const torrentUDPRules = `
udp dport 53 return
@th,64,96 0x64313a6164323a696432303a jump block
@th,64,96 0x64313a7264323a696432303a jump block
@th,64,96 0x000004172710198000000000 jump block
ct state new udp length >= 28 @th,64,8 0x41 @th,72,8 < 3 @th,128,32 0 jump block`
// DHT-запрос «d1:ad2:id20:» и ответ «d1:rd2:id20:» (bencode, ключи отсортированы — «a»/«r» первые);
// UDP-трекер: protocol_id 0x41727101980 + action connect (0); uTP SYN: тип 4, версия 1 (0x41),
// расширение 0–2, timestamp_difference 0 — только в новом потоке, где QUIC начинается с длинного
// заголовка (0xC0+), и не на 53-м порту, чтобы случайный ID DNS-запроса не сошёлся.
const torrentTCPRules = `
@ih,0,128 0x13426974546f7272656e742070726f74 jump block
@ih,0,112 0x474554202f616e6e6f756e63653f jump block`
// «\x13BitTorrent prot» — открытое рукопожатие пиров; «GET /announce?» — HTTP-трекер.
type Torrent struct {
s *Server
opMu sync.Mutex // сериализует nft
mu sync.Mutex
tcp bool
lastError string
cacheAt time.Time
cache map[string]bool
cacheHits float64
}
func newTorrent(s *Server) *Torrent { return &Torrent{s: s} }
func (t *Torrent) Settings() TorrentSettings {
return TorrentSettings{Enabled: t.s.getCfg("torrentBlock", "0") == "1"}
}
func (t *Torrent) script(tcp bool) string {
iface := t.s.iface
var b strings.Builder
// add+delete в одной транзакции: таблица пересоздаётся атомарно, без окна без правил.
b.WriteString("\nadd table inet " + torrentTable + "\ndelete table inet " + torrentTable + "\ntable inet " + torrentTable + " {\n")
b.WriteString(" counter hits {\n }\n")
b.WriteString(" set seen {\n type ipv4_addr\n flags dynamic, timeout\n timeout " + torrentSeenTTL + "\n }\n")
b.WriteString(" chain block {\n ct mark set ct mark or " + torrentMark + "\n update @seen { ip saddr }\n counter name \"hits\"\n drop\n }\n")
b.WriteString(" chain udp_sig {" + torrentUDPRules + "\n }\n")
if tcp {
b.WriteString(" chain tcp_sig {" + torrentTCPRules + "\n }\n")
}
b.WriteString(" chain forward {\n type filter hook forward priority -10; policy accept;\n")
b.WriteString(" ct mark and " + torrentMark + " == " + torrentMark + " drop\n")
b.WriteString(" iifname \"" + iface + "\" meta nfproto ipv4 meta l4proto udp jump udp_sig\n")
if tcp {
b.WriteString(" iifname \"" + iface + "\" meta nfproto ipv4 meta l4proto tcp jump tcp_sig\n")
}
b.WriteString(" }\n}\n")
return b.String()
}
func (t *Torrent) applyNow() error {
if !t.Settings().Enabled {
t.removeNow()
return nil
}
// Сначала полный набор; старый nft/ядро не знают @ih — тогда без TCP, но UDP (DHT, трекеры,
// uTP) всё равно режем: это и есть основная часть.
tcp := true
err := shExecStdin("nft", []string{"-f", "-"}, t.script(true))
if err != nil {
tcp = false
if err2 := shExecStdin("nft", []string{"-f", "-"}, t.script(false)); err2 != nil {
return err2
}
t.s.log.Warn("torrent: TCP-сигнатуры не поддерживаются, блокируем только UDP", "nft", err.Error())
}
t.mu.Lock()
t.tcp, t.lastError, t.cache = tcp, "", nil
t.mu.Unlock()
t.s.log.Info("torrent block applied", "tcp", tcp)
return nil
}
func (t *Torrent) removeNow() {
shExec("nft", "delete", "table", "inet", torrentTable)
t.mu.Lock()
t.tcp, t.cache = false, nil
t.mu.Unlock()
}
func (t *Torrent) noteError(err error) {
t.mu.Lock()
defer t.mu.Unlock()
if m := err.Error(); m != t.lastError {
t.s.log.Warn("torrent: " + m)
t.lastError = m
}
}
// Apply не бросает: без nft awg-ctrl должен работать дальше (как шейпинг).
func (t *Torrent) Apply() {
t.opMu.Lock()
err := t.applyNow()
t.opMu.Unlock()
if err != nil {
t.noteError(err)
}
}
// Update сохраняет и применяет; ошибку отдаёт вызывающему — оператору нужна причина.
func (t *Torrent) Update(st TorrentSettings) error {
v := "0"
if st.Enabled {
v = "1"
}
t.s.setCfg("torrentBlock", v)
t.opMu.Lock()
err := t.applyNow()
t.opMu.Unlock()
if err != nil {
t.mu.Lock()
t.lastError = err.Error()
t.mu.Unlock()
}
return err
}
// read — набор seen и счётчик из ядра. ok=false — таблицы нет.
func (t *Torrent) read() (seen map[string]bool, hits float64, ok bool) {
out, err := shExec("nft", "-j", "list", "set", "inet", torrentTable, "seen")
if err != nil {
return nil, 0, false
}
seen = map[string]bool{}
var doc struct {
Nftables []struct {
Set *struct {
Elem []json.RawMessage `json:"elem"`
} `json:"set"`
Counter *struct {
Packets float64 `json:"packets"`
} `json:"counter"`
} `json:"nftables"`
}
if json.Unmarshal([]byte(out), &doc) == nil {
for _, it := range doc.Nftables {
if it.Set == nil {
continue
}
// Элемент с таймаутом — {"elem": {"val": "10.9.0.2", ...}}, без него — просто строка.
for _, raw := range it.Set.Elem {
var ip string
if json.Unmarshal(raw, &ip) != nil {
var e struct {
Elem struct {
Val string `json:"val"`
} `json:"elem"`
}
if json.Unmarshal(raw, &e) == nil {
ip = e.Elem.Val
}
}
if ip != "" {
seen[ip] = true
}
}
}
}
if out, err := shExec("nft", "-j", "list", "counter", "inet", torrentTable, "hits"); err == nil {
doc.Nftables = nil
if json.Unmarshal([]byte(out), &doc) == nil {
for _, it := range doc.Nftables {
if it.Counter != nil {
hits = it.Counter.Packets
}
}
}
}
return seen, hits, true
}
func (t *Torrent) cached() (map[string]bool, float64, bool) {
if !t.Settings().Enabled {
return nil, 0, false
}
t.mu.Lock()
if t.cache != nil && time.Since(t.cacheAt) < torrentCacheS {
c, h := t.cache, t.cacheHits
t.mu.Unlock()
return c, h, true
}
t.mu.Unlock()
seen, hits, ok := t.read()
if !ok {
return nil, 0, false
}
t.mu.Lock()
t.cache, t.cacheHits, t.cacheAt = seen, hits, time.Now()
t.mu.Unlock()
return seen, hits, true
}
// Seen — пытался ли этот IP открыть торрент-соединение за последние torrentSeenTTL.
func (t *Torrent) Seen(ip string) bool {
seen, _, _ := t.cached()
return seen[ip]
}
// SeenList — IP с меткой «торренты», по порядку; шейпинг держит их в фоновом классе.
func (t *Torrent) SeenList() []string {
seen, _, _ := t.cached()
out := make([]string, 0, len(seen))
for ip := range seen {
out = append(out, ip)
}
sort.Strings(out)
return out
}
func (t *Torrent) Status() TorrentStatus {
st := TorrentStatus{Settings: t.Settings(), Seen: []TorrentPeer{}}
seen, hits, active := t.cached()
t.mu.Lock()
st.TCP, st.Error = t.tcp && active, t.lastError
t.mu.Unlock()
st.Active, st.Hits = active, hits
if len(seen) == 0 {
return st
}
names := map[string]string{}
if rows, err := t.s.db.Query("SELECT name, ip FROM users"); err == nil {
for rows.Next() {
var n, ip string
if rows.Scan(&n, &ip) == nil {
names[ip] = n
}
}
rows.Close()
}
for ip := range seen {
st.Seen = append(st.Seen, TorrentPeer{IP: ip, Name: names[ip]})
}
sort.Slice(st.Seen, func(i, j int) bool { return st.Seen[i].IP < st.Seen[j].IP })
return st
}