host_firewall: per-service new-connection meters per source (/32, /64) and per range (/24, /48) from measured peaks, excess dropped with prefix fw-ratelimit:, replaces the ssh meter

host_firewall/defaults/main.yml

@@ -44,12 +44,36 @@ firewall:
   # in on ifaces pass, the listed ports and types pass, the rest to daddr drops;
   # traffic to other addresses is not filtered.
   forward_guard: {}
-  ssh_rate_limit: "10/minute burst 5 packets"
+  # Limits on new inbound TCP connections per service, applied to the public and geo ports the host serves (input) and to
-  # Kernel log of dropped inbound packets (fw-in-drop: ) and of rejected
+  # DNATed traffic (forward: port forwards and Docker-published ports), after the admin, cluster, VPN, guest and internal
+  # sources are exempted. Each service has a tier per source (IPv4 /32, IPv6 /64) and a tier per range (IPv4 /24, IPv6
+  # /48) at rate * range_factor. The excess drops with the log prefix fw-ratelimit: (log.ratelimit_rate).
+  # name -> {ports: port or list, rate: connections per `per`, per: second|minute (minute), burst: connections (2 * rate),
+  # range_factor_v4, range_factor_v6: override the factors below}; null drops a service, a service is idle unless a port is served.
+  # Values are 3x the highest legitimate rate seen, with a floor of 12 per minute (3x four connections per minute) where
+  # there is no or too little history; the measurement is noted at each service.
+  ratelimit:
+    range_factor:
+      v4: 6
+      v6: 3
+    services:
+      # 14 days of sshd Accepted lines in VictoriaLogs (all hosts, per host and source, /64 and /24 grouped): at most 8 per
+      # minute and 4 per 10 seconds from one address or /64 (Ansible runs), 22 per 10 minutes.
+      ssh:
+        ports: 22
+        rate: 24
+      # 14 days of Traefik access logs of the cluster: distinct client ports of one source within a minute, which bounds the
+      # new connections, at most 51 (IPv4) and 58 (IPv6 /64); 16 per second at most. Applies to the host ports 80 and 443.
+      web:
+        ports: [80, 443]
+        rate: 180
+  # Kernel log of dropped inbound packets (fw-in-drop: ), rate-limited connections (fw-ratelimit: ) and of rejected
   # outbound mail (fw-out-reject: ), read from the journal by CrowdSec. Each
   # rate is one limit shared by all matching packets.
   log:
     in_rate: "5/second burst 30 packets"
+    # Shared by all rate-limited connections (fw-ratelimit: ), one limit.
+    ratelimit_rate: "5/second burst 30 packets"
     out_rate: "6/minute burst 5 packets"
     # Lets the kernel log the nft log statements of container network
     # namespaces (LXC guests) into the host kernel log.

host_firewall/templates/host.nft.j2

@@ -21,6 +21,33 @@ set {{ name }} {
 {% set scrapers = firewall.scrapers | default({}, true) %}
 {% set forward_guard = firewall.forward_guard | default({}, true) | dict2items | rejectattr('value', 'none') | list %}
 {% set blocklists = firewall.spamhaus or firewall.firehol %}
+{% macro port_values(map) -%}
+{{ map | default({}, true) | dict2items | rejectattr('value', 'none') | map(attribute='value') | map('string') | unique | join(' ') }}
+{%- endmacro %}
+{# Per-service limits apply only to ports the host serves: input for the public and geo ports, forward for DNATed traffic (port forwards of the host and Docker-published ports). #}
+{% set served_input = (port_values(firewall.public.tcp) ~ ' ' ~ port_values(firewall.geo.tcp)).split() %}
+{% set served_forward = (port_values(firewall.public.tcp) ~ ' ' ~ port_values(firewall.geo.tcp) ~ ' ' ~ (dnat | map(attribute='tcp') | map('default', {}, true) | map('dict2items') | sum(start=[]) | rejectattr('value', 'none') | map(attribute='value') | map('string') | join(' '))).split() %}
+{% set rl = firewall.ratelimit | default({}, true) %}
+{% set rl_factor = rl.range_factor | default({}, true) %}
+{% set rl_services = [] %}
+{% for svc in rl.services | default({}, true) | dict2items | rejectattr('value', 'none') %}
+{% set ports = [svc.value.ports] | flatten | map('string') | list %}
+{% set per = svc.value.per | default('minute', true) %}
+{% set burst = svc.value.burst | default(svc.value.rate * 2, true) %}
+{% set _ = rl_services.append({
+	'name': svc.key | replace('-', '_'),
+	'ports': ports | join(', '),
+	'rate': svc.value.rate,
+	'burst': burst,
+	'per': per,
+	'f4': svc.value.range_factor_v4 | default(rl_factor.v4 | default(6), true),
+	'f6': svc.value.range_factor_v6 | default(rl_factor.v6 | default(3), true),
+	'input': (ports | select('in', served_input) | list | length) > 0,
+	'forward': (ports | select('in', served_forward) | list | length) > 0,
+}) %}
+{% endfor %}
+{% set rl_input = rl_services | selectattr('input') | list %}
+{% set rl_forward = rl_services | selectattr('forward') | list %}
 
 # Declare-then-delete makes every load one atomic transaction that replaces
 # only these tables; Cilium, Docker, NetBird and CrowdSec tables stay untouched.
@@ -49,12 +76,20 @@ table inet host {
 	{{ addr_set('cluster_v6', 'ipv6_addr', cluster.v6) }}
 	{{ addr_set('scrapers_v4', 'ipv4_addr', scrapers.v4) }}
 	{{ addr_set('scrapers_v6', 'ipv6_addr', scrapers.v6) }}
-	set ssh_meter_v4 {
+{% for svc in rl_services if svc.input or svc.forward %}
+	set rl_{{ svc.name }}_v4 {
 		type ipv4_addr; flags dynamic; timeout 10m; size 65535;
 	}
-	set ssh_meter_v6 {
+	set rl_{{ svc.name }}_v6 {
 		type ipv6_addr; flags dynamic; timeout 10m; size 65535;
 	}
+	set rl_{{ svc.name }}_v4r {
+		type ipv4_addr; flags dynamic; timeout 10m; size 16384;
+	}
+	set rl_{{ svc.name }}_v6r {
+		type ipv6_addr; flags dynamic; timeout 10m; size 16384;
+	}
+{% endfor %}
 
 	chain input {
 		type filter hook input priority filter; policy drop;
@@ -127,6 +162,44 @@ table inet host {
 		limit rate {{ firewall.log.in_rate }} log prefix "fw-in-drop: "
 		drop
 	}
+{% if rl_services | selectattr('input') | list or rl_forward %}
+
+	# Per-service limits on new inbound TCP connections (firewall.ratelimit): per source address (IPv4 /32, IPv6 /64) and per
+	# range (IPv4 /24, IPv6 /48; the per-source rate times range_factor). Admin, cluster, VPN, guest and internal sources
+	# are exempt. The excess drops with its own log prefix, which CrowdSec reads as a ratelimit event and not as a port scan.
+{% macro rl_exempt() %}
+		iifname "wt0" return
+{% for iface in cluster_ifaces + internal_ifaces + guest_ifaces %}
+		iifname "{{ iface }}" return
+{% endfor %}
+		ip saddr @admin_v4 return
+		ip6 saddr @admin_v6 return
+		ip saddr @cluster_v4 return
+		ip6 saddr @cluster_v6 return
+{% endmacro %}
+{% for hook, svcs in [('in', rl_input), ('fwd', rl_forward)] if svcs %}
+	chain ratelimit_{{ hook }} {
+{{ rl_exempt() }}
+{% for svc in svcs %}
+		meta l4proto tcp ct original proto-dst { {{ svc.ports }} } counter jump rl_{{ svc.name }}
+{% endfor %}
+	}
+
+{% endfor %}
+{% for svc in rl_services if svc.input or svc.forward %}
+	chain rl_{{ svc.name }} {
+		update @rl_{{ svc.name }}_v4 { ip saddr limit rate over {{ svc.rate }}/{{ svc.per }} burst {{ svc.burst }} packets } counter jump rl_drop
+		update @rl_{{ svc.name }}_v6 { ip6 saddr & ffff:ffff:ffff:ffff:: limit rate over {{ svc.rate }}/{{ svc.per }} burst {{ svc.burst }} packets } counter jump rl_drop
+		update @rl_{{ svc.name }}_v4r { ip saddr & 255.255.255.0 limit rate over {{ svc.rate * svc.f4 }}/{{ svc.per }} burst {{ svc.burst * svc.f4 }} packets } counter jump rl_drop
+		update @rl_{{ svc.name }}_v6r { ip6 saddr & ffff:ffff:ffff:: limit rate over {{ svc.rate * svc.f6 }}/{{ svc.per }} burst {{ svc.burst * svc.f6 }} packets } counter jump rl_drop
+	}
+
+{% endfor %}
+	chain rl_drop {
+		limit rate {{ firewall.log.ratelimit_rate }} log prefix "fw-ratelimit: "
+		drop
+	}
+{% endif %}
 {% if firewall.spamhaus %}
 
 	# Own limits, so a listed scanner cannot use up the budget of fw-in-drop.
@@ -204,6 +277,9 @@ table inet host {
 	}
 
 	chain public {
+{% if rl_input %}
+		ct state new jump ratelimit_in
+{% endif %}
 {% if list(firewall.public.tcp) %}
 		tcp dport { {{ list(firewall.public.tcp) }} } accept
 {% endif %}
@@ -232,18 +308,20 @@ table inet host {
 {% endif %}
 		ip saddr @admin_v4 accept
 		ip6 saddr @admin_v6 accept
-		tcp dport 22 ct state new update @ssh_meter_v4 { ip saddr limit rate over {{ firewall.ssh_rate_limit }} } jump drop_log
-		# One IPv6 client usually holds a whole /64.
-		tcp dport 22 ct state new update @ssh_meter_v6 { ip6 saddr & ffff:ffff:ffff:ffff:: limit rate over {{ firewall.ssh_rate_limit }} } jump drop_log
 {% if firewall.geo_countries %}
 		ip saddr @geo_v4 accept
 		ip6 saddr @geo_v6 accept
 {% endif %}
 	}
-{% if forward_guard or blocklists or firewall.scanners %}
+{% if forward_guard or blocklists or firewall.scanners or rl_forward %}
 
 	chain forward {
 		type filter hook forward priority filter; policy accept;
+{% if rl_forward %}
+
+		# Docker-published ports and port forwards arrive here after DNAT; pod and guest egress is not DNATed.
+		ct status dnat ct state new jump ratelimit_fwd
+{% endif %}
 {% if firewall.spamhaus %}
 
 		ip saddr @spamhaus_v4 jump spamhaus_in