#!/bin/sh
# Copyright (C) 2024 Ycarus (Yannick Chabanois) <ycarus@zugaina.org> for OpenMPTCProuter
#
# nDPId consumer for omr-bypass.
# Reads flow detection events from the ndpisrvd distributor socket and adds
# destination IPs to the appropriate nftables named sets so that traffic
# classified by nDPId is bypassed according to omr-bypass protocol rules.
#
# Usage: ndpid-consumer <proto_list>
#   proto_list: space-separated entries of the form
#               proto:intf:intfid:dscp:family:tcpudp as written by
#               _bypass_ndpid_collect_proto in omr-bypass-nft.
#               intf and dscp are independent and either may be empty: intf
#               drives the routing-bypass fwmark set, dscp drives the
#               omr_dst_bypass_dscp_<class> set -- a rule can populate one, the
#               other, or both from the same detected flow.

PIDFILE="/var/run/omr-bypass-ndpid.pid"
SOCK="/var/run/ndpid/distributor.sock"

proto_list="$1"
[ -z "$proto_list" ] && exit 1

# Everything below comes from the stream nobody-owned nDPId feeds, or goes
# into an nft command line: no globbing of either
set -f

# _valid_ip4 / _valid_ip6 <address>: as in omr-bypass-nft. nft joins its
# arguments into one command, so a dst_ip "1.2.3.4 }; flush ruleset" sent
# by nDPId was run as root.
_valid_ip4() {
	local IFS=. __o
	case "$1" in ""|*[!0-9.]*|.*|*.|*..*) return 1 ;; esac
	set -- $1
	[ $# -eq 4 ] || return 1
	for __o in "$@"; do
		case "$__o" in ????*|0?*) return 1 ;; esac
		[ "$__o" -le 255 ] || return 1
	done
}

_valid_ip6() {
	local __a="$1" __g __n=0 IFS=:
	case "$__a" in *:*) ;; *) return 1 ;; esac
	case "$__a" in *[!0-9A-Fa-f:.]*|*:::*|*::*::*|:[!:]*|*[!:]:) return 1 ;; esac
	# an embedded IPv4 address counts as two groups
	case "$__a" in
		*.*)
			_valid_ip4 "${__a##*:}" || return 1
			__a="${__a%:*}:0:0"
			;;
	esac
	set -- $__a
	for __g in "$@"; do
		[ -n "$__g" ] || continue
		case "$__g" in *.*|?????*) return 1 ;; esac
		__n=$((__n + 1))
	done
	case "$__a" in
		*::*) [ "$__n" -le 7 ] ;;
		*) [ "$__n" -eq 8 ] ;;
	esac
}

# The set names are made of the proto, interface and DSCP class of each
# entry. The proto keeps what omr-bypass-nft's _rule_proto allows, as
# nDPI names have dashes (JSON-RPC, NAT-PMP) and so do nft set names.
_list=""
for entry in $proto_list; do
	_f="${entry}:::::"
	bp="${_f%%:*}"; _f="${_f#*:}"
	intf="${_f%%:*}"; _f="${_f#*:}"
	_f="${_f#*:}"
	dscp="${_f%%:*}"; _f="${_f#*:}"
	_f="${_f#*:}"
	tcpudp="${_f%%:*}"
	case "$bp" in
		""|*[!A-Za-z0-9_.-]*) entry="" ;;
	esac
	case "$intf$dscp" in
		*[!A-Za-z0-9_]*) entry="" ;;
	esac
	case "$tcpudp" in
		""|tcp|udp) ;;
		*) entry="" ;;
	esac
	if [ -z "$entry" ]; then
		logger -t omr-bypass "nDPId consumer: invalid entry '$bp:$intf:$dscp:$tcpudp' ignored"
		continue
	fi
	_list="$_list $entry"
done
proto_list="${_list# }"
[ -z "$proto_list" ] && exit 1

echo $$ > "$PIDFILE"

# ndpisrvd exposes two distributor endpoints:
#   - TCP (127.0.0.1:7000 by default) — works with GNU netcat and BusyBox nc
#   - Unix socket ($SOCK)             — requires OpenBSD netcat or socat
# Prefer TCP so the consumer works on OpenWrt with GNU netcat.
TCP_HOST=$(uci -q get ndpid.distributor.tcp_address 2>/dev/null)
TCP_PORT=$(uci -q get ndpid.distributor.tcp_port 2>/dev/null)
TCP_HOST=${TCP_HOST:-127.0.0.1}
TCP_PORT=${TCP_PORT:-7000}

# nc gets a stdin that never ends and is not a pipeline: a FIFO opened
# read-write, which this script holds and never writes to. With
# "tail -f /dev/null | nc" the shell also waited for tail, which never
# writes and so never gets SIGPIPE: once ndpisrvd restarted and nc exited,
# the consumer hung for good and nDPI bypass stopped. Now nc's exit ends the
# read loop below and the distributor is connected to again. (GNU netcat
# 0.7.1 keeps reading the socket after a stdin EOF, but an nc that quits
# then would read nothing at all.)
STDIN_FIFO="${PIDFILE%.pid}.stdin"
rm -f "$STDIN_FIFO"
mkfifo "$STDIN_FIFO" || exit 1
exec 3<>"$STDIN_FIFO"
rm -f "$STDIN_FIFO"

_nc_distributor() {
	if nc "$TCP_HOST" "$TCP_PORT" <&3 2>/dev/null; then
		return
	fi
	[ ! -S "$SOCK" ] && return
	nc -U "$SOCK" <&3 2>/dev/null
}

while true; do
	_nc_distributor | while IFS= read -r line; do
		# Most events are not detections (new, idle, update, end, packet,
		# daemon status): skip them before forking jsonfilter for each.
		case "$line" in
			*'"flow_event_name":"detect'*) ;;
			*) continue ;;
		esac
		# nDPId frames are prefixed with a 5-character hex length field
		json="${line:5}"
		[ -z "$json" ] && continue

		event=$(echo "$json" | jsonfilter -q -e '@.flow_event_name' 2>/dev/null)
		[ "$event" != "detected" ] && [ "$event" != "detection-update" ] && continue

		proto_str=$(echo "$json" | jsonfilter -q -e '@.ndpi.proto' 2>/dev/null)
		[ -z "$proto_str" ] && continue

		# ndpisrvd uses "dst_ip" (not "l3_dst4"/"l3_dst6"); the address
		# itself tells its family
		dst_ip=$(echo "$json" | jsonfilter -q -e '@.dst_ip' 2>/dev/null)
		if _valid_ip4 "$dst_ip"; then
			is_v6=0
		elif _valid_ip6 "$dst_ip"; then
			is_v6=1
		else
			continue
		fi

		# Normalize nDPI proto string (e.g. "TLS.Google" -> app="google", base="tls")
		app=$(echo "$proto_str" | tr '.' '\n' | tail -1 | tr '[:upper:]' '[:lower:]')
		base=$(echo "$proto_str" | tr '.' '\n' | head -1 | tr '[:upper:]' '[:lower:]')

		for entry in $proto_list; do
			# proto:intf:intfid:dscp:family:tcpudp, any field but proto
			# may be empty and an older list has fewer fields: pad, then
			# split
			_f="${entry}:::::"
			bp="${_f%%:*}"; _f="${_f#*:}"
			intf="${_f%%:*}"; _f="${_f#*:}"
			_f="${_f#*:}"
			dscp="${_f%%:*}"; _f="${_f#*:}"
			family="${_f%%:*}"; _f="${_f#*:}"
			tcpudp="${_f%%:*}"
			case "$family:$is_v6" in
				ipv4:1|ipv6:0) continue ;;
			esac
			# a TCP or UDP only rule: the interface's set of that
			# protocol, the one its MARK rule matches ("all" has a set
			# per protocol already, the server VPN none per protocol)
			_p=""
			case "$intf:$tcpudp" in
				all:*|srv_vpn1:*) ;;
				*:tcp|*:udp) _p="_$tcpudp" ;;
			esac
			bp_lower=$(echo "$bp" | tr '[:upper:]' '[:lower:]')
			if [ "$bp_lower" = "$app" ] || [ "$bp_lower" = "$base" ] || \
			   echo "$proto_str" | grep -qi "^${bp}$" || \
			   echo "$proto_str" | grep -qi "\.${bp}$" || \
			   echo "$proto_str" | grep -qi "^${bp}\."; then
				# Routing-bypass fwmark and DSCP marking are independent --
				# a rule can have an interface, a dscp class, or both (same
				# convention as _bypass_asn/_dscp_ip for the static lists).
				if [ "$is_v6" = "1" ]; then
					if [ "$intf" = "all" ]; then
						nft add element inet fw4 bypass6_${bp} { "$dst_ip" } 2>/dev/null
					elif [ -n "$intf" ]; then
						nft add element inet fw4 omr_dst_bypass_${intf}_6${_p} { "$dst_ip" } 2>/dev/null
					fi
					[ -n "$dscp" ] && nft add element inet fw4 omr_dst_bypass_dscp_${dscp}_6 { "$dst_ip" } 2>/dev/null
				else
					if [ "$intf" = "all" ]; then
						nft add element inet fw4 bypass_${bp} { "$dst_ip" } 2>/dev/null
					elif [ -n "$intf" ]; then
						nft add element inet fw4 omr_dst_bypass_${intf}_4${_p} { "$dst_ip" } 2>/dev/null
					fi
					[ -n "$dscp" ] && nft add element inet fw4 omr_dst_bypass_dscp_${dscp}_4 { "$dst_ip" } 2>/dev/null
				fi
			fi
		done
	done
	sleep 2
done
