Copilot commented on code in PR #13586: URL: https://github.com/apache/trafficserver/pull/13586#discussion_r3906580459
########## tests/gold_tests/pluginTest/abuse_shield/abuse_shield.test.py: ########## @@ -0,0 +1,1380 @@ +""" +Verify abuse_shield plugin functionality. +""" +import hashlib +import sys + +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +Test.Summary = ''' +Verify abuse_shield plugin initialization and message handling via traffic_ctl. +''' + +Test.SkipUnless(Condition.PluginExists('abuse_shield.so'), Condition.PluginExists('jax_fingerprint.so')) + + +class AbuseShieldMessageTest: + """Verify abuse_shield plugin message handling.""" + + def __init__(self): + """Set up the test environment and run all test scenarios.""" + self._setup_ts() + self._test_plugin_initialization() + self._test_disable_plugin() + self._test_enable_plugin() + self._test_dump_command() + self._test_stats_command() + self._test_reset_command() + self._test_trusted_command() + self._test_trusted_bypass() + self._test_reload_command() + + def _setup_ts(self) -> None: + """Configure ATS with the abuse_shield plugin.""" + self._ts = Test.MakeATSProcess("ts") + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + }) + + # Create the plugin config file. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + f''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + + trusted_ips_file: {self._ts.Variables.CONFIGDIR}/abuse_shield_trusted.yaml + +rules: + - name: "test_h2_error_rule" + filter: + max_h2_error_rate: 5 + action: [log, block] + + - name: "test_request_rule" + filter: + max_req_rate: 10 + action: [log, block, close] + +enabled: true +'''.strip().split('\n')) + + # Create trusted IPs file (YAML format). + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield_trusted.yaml", id="trusted_yaml", typename="ats:config") + self._ts.Disk.trusted_yaml.AddLines(''' +trusted_ips: + - 127.0.0.1 + - "::1" + - "::ffff:127.0.0.1" +'''.strip().split('\n')) + + # Configure abuse_shield plugin. + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + # Verify the plugin loads. The plugin logs to diags.log via TSError. + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded successfully.") + + def _test_plugin_initialization(self) -> None: + """Verify the plugin starts with configured values.""" + tr = Test.AddTestRun("Verify plugin starts with configured values.") + tr.Processes.Default.Command = "echo verifying plugin starts with configured values" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # The "Plugin initialized" NOTE message confirms the plugin loaded with correct config. + # (The debug-level "Created 3 IP trackers" message only appears when debug output is enabled.) + + def _test_disable_plugin(self) -> None: + """Verify the 'enabled' setting can be changed via traffic_ctl.""" + tr = Test.AddTestRun("Verify changing 'enabled' via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.enabled 0" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the enabled change.") + tr.Processes.Default.Command = "echo awaiting enabled change" + tr.Processes.Default.ReturnCode = 0 + await_enabled = tr.Processes.Process('await_enabled', 'sleep 30') + await_enabled.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Plugin disabled") + tr.Processes.Default.StartBefore(await_enabled) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Plugin disabled", "Verify abuse_shield received the disabled command.") + + def _test_enable_plugin(self) -> None: + """Re-enable the plugin via traffic_ctl.""" + tr = Test.AddTestRun("Re-enable the plugin via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.enabled 1" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the re-enable.") + tr.Processes.Default.Command = "echo awaiting re-enable" + tr.Processes.Default.ReturnCode = 0 + await_reenable = tr.Processes.Process('await_reenable', 'sleep 30') + await_reenable.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Plugin enabled") + tr.Processes.Default.StartBefore(await_reenable) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Plugin enabled", "Verify abuse_shield received the enabled command.") + + def _test_dump_command(self) -> None: + """Verify dump command via traffic_ctl.""" + tr = Test.AddTestRun("Verify dump command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.dump" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the dump output.") + tr.Processes.Default.Command = "echo awaiting dump output" + tr.Processes.Default.ReturnCode = 0 + await_dump = tr.Processes.Process('await_dump', 'sleep 30') + await_dump.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "abuse_shield dump") + tr.Processes.Default.StartBefore(await_dump) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "abuse_shield.*Dump:", "Verify abuse_shield dump command works.") + + def _test_reload_command(self) -> None: + """Verify reload command via traffic_ctl.""" + tr = Test.AddTestRun("Verify reload command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.reload" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the reload.") + tr.Processes.Default.Command = "echo awaiting reload" + tr.Processes.Default.ReturnCode = 0 + await_reload = tr.Processes.Process('await_reload', 'sleep 30') + await_reload.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Configuration reloaded successfully") + tr.Processes.Default.StartBefore(await_reload) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Configuration reloaded successfully", "Verify abuse_shield reload command works.") + + def _test_stats_command(self) -> None: + """Verify the stats command synchronizes table metrics.""" + tr = Test.AddTestRun("Verify stats command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.stats" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression("Stats synced", "Verify abuse_shield stats command works.") + + def _test_reset_command(self) -> None: + """Verify the reset command clears plugin metrics.""" + tr = Test.AddTestRun("Verify reset command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.reset" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression("Metrics reset", "Verify abuse_shield reset command works.") + + def _test_trusted_command(self) -> None: + """Verify the trusted command reports configured bypass ranges.""" + tr = Test.AddTestRun("Verify trusted command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.trusted" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r"Trusted IP ranges \(3 total\)", "Verify abuse_shield trusted command reports all ranges.") + + def _test_trusted_bypass(self) -> None: + """Verify trusted traffic bypasses request-rate enforcement.""" + tr = Test.AddTestRun("Verify trusted IP bypasses rate enforcement.") + tr.Processes.Default.Command = ( + f'seq 1 30 | xargs -P 30 -I {{}} ' + f'curl -s -o /dev/null http://127.0.0.1:{self._ts.Variables.port}/') + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ExcludesExpression( + r'Rule "test_request_rule" matched', "Verify trusted requests did not trigger the configured rule.") + + tr = Test.AddTestRun("Verify trusted traffic did not block the client.") + tr.Processes.Default.Command = "traffic_ctl metric get abuse_shield.actions.blocked" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.Processes.Default.Streams.stdout = Testers.ContainsExpression( + r"abuse_shield.actions.blocked\s+0", "Verify no block action was recorded for trusted traffic.") + tr.StillRunningAfter = self._ts + + +class AbuseShieldRateLimitTest: + """Verify abuse_shield plugin can detect and block request rate floods. + + This test sends HTTP/2 requests at a rate exceeding the configured + max_req_rate threshold and verifies that the plugin detects this and + blocks the offending IP. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run rate limit test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_h2_error_rate_exceeded() + self._test_rate_limit_exceeded() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin{AbuseShieldRateLimitTest._server_counter}' + AbuseShieldRateLimitTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + # Add a simple response for GET requests. + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with TLS and the abuse_shield plugin for rate limiting.""" + name = f'ts_rate{AbuseShieldRateLimitTest._ts_counter}' + AbuseShieldRateLimitTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + # Configure SSL for ATS. + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + + # Remap to the origin server. + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + # Configure records. + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.http.insert_response_via_str': 2, + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create the plugin config file with a low max_req_rate for testing. + # With max_req_rate: 20, sending 50 requests should trigger the rule. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + + log_interval_sec: 0 + +rules: + - name: "req_rate_flood" + filter: + max_req_rate: 20 + action: [log, block] + + - name: "h2_error_flood" + filter: + max_h2_error_rate: 2 + action: [log] + +enabled: true +'''.strip().split('\n')) + + # Configure abuse_shield plugin. + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + # Verify the plugin loads. + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded with rate limit rule.") + + def _test_h2_error_rate_exceeded(self) -> None: + """Reset H2 streams to exercise the H2 error-rate tracker.""" + tr = Test.AddTestRun("Send H2 CANCEL errors to trigger H2 error rate limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 10 --rate 100 --path / --reset-streams') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "h2_error_flood" matched for IP=.*actions=\[log\]', "Verify H2 CANCEL errors triggered the H2 error-rate rule.") + + def _test_rate_limit_exceeded(self) -> None: + """Send requests exceeding the rate threshold and verify blocking. + + This test sends 50 requests at 100 req/sec, which should exceed the + max_req_rate of 20 and trigger the req_rate_flood rule. + """ + tr = Test.AddTestRun("Send excessive H2 requests to trigger rate limit") + + # Send 50 requests at high rate - this should exceed max_req_rate of 20. + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 50 --rate 100 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + + # Verify the rate limit rule was triggered and block action was taken. + # The plugin logs via TSError to diags.log when a rule matches. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "req_rate_flood" matched for IP=.*actions=\[log,block\]', + "Verify the req_rate_flood rule was triggered and block action was taken.") + + +class AbuseShieldConnRateTest: + """Verify abuse_shield plugin can detect and block connection rate floods. + + This test opens many connections rapidly to exceed the configured + max_conn_rate threshold and verifies that the plugin detects this and + blocks the offending IP. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run connection rate test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_conn_rate_exceeded() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_conn{AbuseShieldConnRateTest._server_counter}' + AbuseShieldConnRateTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with TLS and the abuse_shield plugin for connection rate limiting.""" + name = f'ts_conn{AbuseShieldConnRateTest._ts_counter}' + AbuseShieldConnRateTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create the plugin config file with a low max_conn_rate for testing. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "conn_rate_flood" + filter: + max_conn_rate: 5 + action: [log, block] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 1 rules", + "Verify the abuse_shield plugin loaded with connection rate limit rule.") + + def _test_conn_rate_exceeded(self) -> None: + """Open idle TLS connections to enforce before ClientHello.""" + tr = Test.AddTestRun("Trigger connection rule with idle TLS connections") + + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/idle_connections.py ' + f'--port {self._ts.Variables.ssl_port} --count 30') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # Verify the connection rate rule was triggered and block action was taken. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "conn_rate_flood" matched for IP=.*actions=\[log,block\]', + "Verify the conn_rate_flood rule was triggered and block action was taken.") + + +class AbuseShieldRateLimitedIpRateLimitTest: + """Verify rate-limited IP rules replace ordinary rules for the rate-limited metric.""" + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run rate-limited IP request-rate scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_rate_limited_ip_uses_rate_limited_rule() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_rate_limited{AbuseShieldRateLimitedIpRateLimitTest._server_counter}' + AbuseShieldRateLimitedIpRateLimitTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with an ordinary request rule and a rate-limited IP request rule.""" + name = f'ts_rate_limited{AbuseShieldRateLimitedIpRateLimitTest._ts_counter}' + AbuseShieldRateLimitedIpRateLimitTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + self._ts.Disk.File( + self._ts.Variables.CONFIGDIR + "/abuse_shield_rate_limited.yaml", id="rate_limited_yaml", typename="ats:config") + self._ts.Disk.rate_limited_yaml.AddLines( + ''' +rate_limited_ips: + - 127.0.0.1 + - "::1" + - "::ffff:127.0.0.1" +'''.strip().split('\n')) + + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + f''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "ordinary_req" + filter: + max_req_rate: 5 + action: [log, block] + + - name: "rate_limited_req" + filter: + max_req_rate: 100 + rate_limited_ips_file: {self._ts.Variables.CONFIGDIR}/abuse_shield_rate_limited.yaml + action: [log] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded with rate-limited IP rule.") + + def _test_rate_limited_ip_uses_rate_limited_rule(self) -> None: + """Send traffic that crosses ordinary then rate-limited request thresholds.""" + tr = Test.AddTestRun("Send rate-limited IP traffic above ordinary request limit but below rate-limited limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 30 --rate 50 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Send rate-limited IP traffic above rate-limited request limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 250 --rate 1000 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + Review Comment: This method has two consecutive TestRuns, but only the first run starts the origin server and only ATS is listed in `StillRunningAfter`. In AuTest, processes not listed in `StillRunningAfter` can be stopped between runs, so the origin may be torn down before the second request burst, making the test sensitive to cache behavior and potentially flaky. Keep the origin running across both runs (see e.g. tests/gold_tests/h2/h2enable.test.py:66-95). ########## tests/gold_tests/pluginTest/abuse_shield/abuse_shield.test.py: ########## @@ -0,0 +1,1380 @@ +""" +Verify abuse_shield plugin functionality. +""" +import hashlib +import sys + +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +Test.Summary = ''' +Verify abuse_shield plugin initialization and message handling via traffic_ctl. +''' + +Test.SkipUnless(Condition.PluginExists('abuse_shield.so'), Condition.PluginExists('jax_fingerprint.so')) + + +class AbuseShieldMessageTest: + """Verify abuse_shield plugin message handling.""" + + def __init__(self): + """Set up the test environment and run all test scenarios.""" + self._setup_ts() + self._test_plugin_initialization() + self._test_disable_plugin() + self._test_enable_plugin() + self._test_dump_command() + self._test_stats_command() + self._test_reset_command() + self._test_trusted_command() + self._test_trusted_bypass() + self._test_reload_command() + + def _setup_ts(self) -> None: + """Configure ATS with the abuse_shield plugin.""" + self._ts = Test.MakeATSProcess("ts") + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + }) + + # Create the plugin config file. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + f''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + + trusted_ips_file: {self._ts.Variables.CONFIGDIR}/abuse_shield_trusted.yaml + +rules: + - name: "test_h2_error_rule" + filter: + max_h2_error_rate: 5 + action: [log, block] + + - name: "test_request_rule" + filter: + max_req_rate: 10 + action: [log, block, close] + +enabled: true +'''.strip().split('\n')) + + # Create trusted IPs file (YAML format). + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield_trusted.yaml", id="trusted_yaml", typename="ats:config") + self._ts.Disk.trusted_yaml.AddLines(''' +trusted_ips: + - 127.0.0.1 + - "::1" + - "::ffff:127.0.0.1" +'''.strip().split('\n')) + + # Configure abuse_shield plugin. + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + # Verify the plugin loads. The plugin logs to diags.log via TSError. + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded successfully.") + + def _test_plugin_initialization(self) -> None: + """Verify the plugin starts with configured values.""" + tr = Test.AddTestRun("Verify plugin starts with configured values.") + tr.Processes.Default.Command = "echo verifying plugin starts with configured values" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # The "Plugin initialized" NOTE message confirms the plugin loaded with correct config. + # (The debug-level "Created 3 IP trackers" message only appears when debug output is enabled.) + + def _test_disable_plugin(self) -> None: + """Verify the 'enabled' setting can be changed via traffic_ctl.""" + tr = Test.AddTestRun("Verify changing 'enabled' via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.enabled 0" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the enabled change.") + tr.Processes.Default.Command = "echo awaiting enabled change" + tr.Processes.Default.ReturnCode = 0 + await_enabled = tr.Processes.Process('await_enabled', 'sleep 30') + await_enabled.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Plugin disabled") + tr.Processes.Default.StartBefore(await_enabled) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Plugin disabled", "Verify abuse_shield received the disabled command.") + + def _test_enable_plugin(self) -> None: + """Re-enable the plugin via traffic_ctl.""" + tr = Test.AddTestRun("Re-enable the plugin via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.enabled 1" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the re-enable.") + tr.Processes.Default.Command = "echo awaiting re-enable" + tr.Processes.Default.ReturnCode = 0 + await_reenable = tr.Processes.Process('await_reenable', 'sleep 30') + await_reenable.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Plugin enabled") + tr.Processes.Default.StartBefore(await_reenable) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Plugin enabled", "Verify abuse_shield received the enabled command.") + + def _test_dump_command(self) -> None: + """Verify dump command via traffic_ctl.""" + tr = Test.AddTestRun("Verify dump command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.dump" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the dump output.") + tr.Processes.Default.Command = "echo awaiting dump output" + tr.Processes.Default.ReturnCode = 0 + await_dump = tr.Processes.Process('await_dump', 'sleep 30') + await_dump.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "abuse_shield dump") + tr.Processes.Default.StartBefore(await_dump) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "abuse_shield.*Dump:", "Verify abuse_shield dump command works.") + + def _test_reload_command(self) -> None: + """Verify reload command via traffic_ctl.""" + tr = Test.AddTestRun("Verify reload command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.reload" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Await the reload.") + tr.Processes.Default.Command = "echo awaiting reload" + tr.Processes.Default.ReturnCode = 0 + await_reload = tr.Processes.Process('await_reload', 'sleep 30') + await_reload.Ready = When.FileContains(self._ts.Disk.diags_log.Name, "Configuration reloaded successfully") + tr.Processes.Default.StartBefore(await_reload) + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + "Configuration reloaded successfully", "Verify abuse_shield reload command works.") + + def _test_stats_command(self) -> None: + """Verify the stats command synchronizes table metrics.""" + tr = Test.AddTestRun("Verify stats command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.stats" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression("Stats synced", "Verify abuse_shield stats command works.") + + def _test_reset_command(self) -> None: + """Verify the reset command clears plugin metrics.""" + tr = Test.AddTestRun("Verify reset command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.reset" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression("Metrics reset", "Verify abuse_shield reset command works.") + + def _test_trusted_command(self) -> None: + """Verify the trusted command reports configured bypass ranges.""" + tr = Test.AddTestRun("Verify trusted command via traffic_ctl.") + tr.Processes.Default.Command = "traffic_ctl plugin msg abuse_shield.trusted" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r"Trusted IP ranges \(3 total\)", "Verify abuse_shield trusted command reports all ranges.") + + def _test_trusted_bypass(self) -> None: + """Verify trusted traffic bypasses request-rate enforcement.""" + tr = Test.AddTestRun("Verify trusted IP bypasses rate enforcement.") + tr.Processes.Default.Command = ( + f'seq 1 30 | xargs -P 30 -I {{}} ' + f'curl -s -o /dev/null http://127.0.0.1:{self._ts.Variables.port}/') + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ExcludesExpression( + r'Rule "test_request_rule" matched', "Verify trusted requests did not trigger the configured rule.") + + tr = Test.AddTestRun("Verify trusted traffic did not block the client.") + tr.Processes.Default.Command = "traffic_ctl metric get abuse_shield.actions.blocked" + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.Env = self._ts.Env + tr.Processes.Default.Streams.stdout = Testers.ContainsExpression( + r"abuse_shield.actions.blocked\s+0", "Verify no block action was recorded for trusted traffic.") + tr.StillRunningAfter = self._ts + + +class AbuseShieldRateLimitTest: + """Verify abuse_shield plugin can detect and block request rate floods. + + This test sends HTTP/2 requests at a rate exceeding the configured + max_req_rate threshold and verifies that the plugin detects this and + blocks the offending IP. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run rate limit test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_h2_error_rate_exceeded() + self._test_rate_limit_exceeded() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin{AbuseShieldRateLimitTest._server_counter}' + AbuseShieldRateLimitTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + # Add a simple response for GET requests. + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with TLS and the abuse_shield plugin for rate limiting.""" + name = f'ts_rate{AbuseShieldRateLimitTest._ts_counter}' + AbuseShieldRateLimitTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + # Configure SSL for ATS. + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + + # Remap to the origin server. + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + # Configure records. + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.http.insert_response_via_str': 2, + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create the plugin config file with a low max_req_rate for testing. + # With max_req_rate: 20, sending 50 requests should trigger the rule. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + + log_interval_sec: 0 + +rules: + - name: "req_rate_flood" + filter: + max_req_rate: 20 + action: [log, block] + + - name: "h2_error_flood" + filter: + max_h2_error_rate: 2 + action: [log] + +enabled: true +'''.strip().split('\n')) + + # Configure abuse_shield plugin. + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + # Verify the plugin loads. + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded with rate limit rule.") + + def _test_h2_error_rate_exceeded(self) -> None: + """Reset H2 streams to exercise the H2 error-rate tracker.""" + tr = Test.AddTestRun("Send H2 CANCEL errors to trigger H2 error rate limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 10 --rate 100 --path / --reset-streams') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "h2_error_flood" matched for IP=.*actions=\[log\]', "Verify H2 CANCEL errors triggered the H2 error-rate rule.") + + def _test_rate_limit_exceeded(self) -> None: + """Send requests exceeding the rate threshold and verify blocking. + + This test sends 50 requests at 100 req/sec, which should exceed the + max_req_rate of 20 and trigger the req_rate_flood rule. + """ + tr = Test.AddTestRun("Send excessive H2 requests to trigger rate limit") + + # Send 50 requests at high rate - this should exceed max_req_rate of 20. + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 50 --rate 100 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + + # Verify the rate limit rule was triggered and block action was taken. + # The plugin logs via TSError to diags.log when a rule matches. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "req_rate_flood" matched for IP=.*actions=\[log,block\]', + "Verify the req_rate_flood rule was triggered and block action was taken.") + + +class AbuseShieldConnRateTest: + """Verify abuse_shield plugin can detect and block connection rate floods. + + This test opens many connections rapidly to exceed the configured + max_conn_rate threshold and verifies that the plugin detects this and + blocks the offending IP. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run connection rate test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_conn_rate_exceeded() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_conn{AbuseShieldConnRateTest._server_counter}' + AbuseShieldConnRateTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with TLS and the abuse_shield plugin for connection rate limiting.""" + name = f'ts_conn{AbuseShieldConnRateTest._ts_counter}' + AbuseShieldConnRateTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create the plugin config file with a low max_conn_rate for testing. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "conn_rate_flood" + filter: + max_conn_rate: 5 + action: [log, block] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 1 rules", + "Verify the abuse_shield plugin loaded with connection rate limit rule.") + + def _test_conn_rate_exceeded(self) -> None: + """Open idle TLS connections to enforce before ClientHello.""" + tr = Test.AddTestRun("Trigger connection rule with idle TLS connections") + + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/idle_connections.py ' + f'--port {self._ts.Variables.ssl_port} --count 30') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # Verify the connection rate rule was triggered and block action was taken. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "conn_rate_flood" matched for IP=.*actions=\[log,block\]', + "Verify the conn_rate_flood rule was triggered and block action was taken.") + + +class AbuseShieldRateLimitedIpRateLimitTest: + """Verify rate-limited IP rules replace ordinary rules for the rate-limited metric.""" + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run rate-limited IP request-rate scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_rate_limited_ip_uses_rate_limited_rule() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_rate_limited{AbuseShieldRateLimitedIpRateLimitTest._server_counter}' + AbuseShieldRateLimitedIpRateLimitTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with an ordinary request rule and a rate-limited IP request rule.""" + name = f'ts_rate_limited{AbuseShieldRateLimitedIpRateLimitTest._ts_counter}' + AbuseShieldRateLimitedIpRateLimitTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + self._ts.Disk.File( + self._ts.Variables.CONFIGDIR + "/abuse_shield_rate_limited.yaml", id="rate_limited_yaml", typename="ats:config") + self._ts.Disk.rate_limited_yaml.AddLines( + ''' +rate_limited_ips: + - 127.0.0.1 + - "::1" + - "::ffff:127.0.0.1" +'''.strip().split('\n')) + + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + f''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "ordinary_req" + filter: + max_req_rate: 5 + action: [log, block] + + - name: "rate_limited_req" + filter: + max_req_rate: 100 + rate_limited_ips_file: {self._ts.Variables.CONFIGDIR}/abuse_shield_rate_limited.yaml + action: [log] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded with rate-limited IP rule.") + + def _test_rate_limited_ip_uses_rate_limited_rule(self) -> None: + """Send traffic that crosses ordinary then rate-limited request thresholds.""" + tr = Test.AddTestRun("Send rate-limited IP traffic above ordinary request limit but below rate-limited limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 30 --rate 50 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + tr = Test.AddTestRun("Send rate-limited IP traffic above rate-limited request limit") + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 250 --rate 1000 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ExcludesExpression( + r'Rule "ordinary_req" matched', "Verify the ordinary request rule did not match rate-limited IP traffic.") + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "rate_limited_req" matched for IP=.*actions=\[log\]', + "Verify the rate-limited request rule matched rate-limited IP traffic.") + + +class AbuseShieldHTTPBlockTest: + """Verify abuse_shield plugin blocks plain HTTP connections via SSN_START hook. + + This test sends HTTP/1.1 requests (not HTTPS) to verify that blocking + works for plain HTTP connections using the SSN_START hook. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run HTTP blocking test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_http_rate_limit() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_http{AbuseShieldHTTPBlockTest._server_counter}' + AbuseShieldHTTPBlockTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS for plain HTTP (no TLS) with the abuse_shield plugin.""" + name = f'ts_http{AbuseShieldHTTPBlockTest._ts_counter}' + AbuseShieldHTTPBlockTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=False, enable_cache=True) + + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + }) + + # Create the plugin config file with a low max_req_rate for testing. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "http_conn_rate_flood" + filter: + max_conn_rate: 10 + action: [log, block] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 1 rules", + "Verify the abuse_shield plugin loaded with HTTP rate limit rule.") + + def _test_http_rate_limit(self) -> None: + """Open plain HTTP sessions exceeding the connection threshold.""" + tr = Test.AddTestRun("Send plain HTTP connections to trigger connection limit") + + # Each curl opens a plain HTTP session, exercising TS_HTTP_SSN_START. + # xargs returns 123 when at least one curl is rejected by the block. + client_cmd = (f'seq 1 50 | xargs -P 50 -I {{}} ' + f'curl --max-time 5 -s http://127.0.0.1:{self._ts.Variables.port}/') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 123 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # Verify the rate limit rule was triggered for HTTP and block action was taken. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "http_conn_rate_flood" matched for IP=.*actions=\[log,block\]', + "Verify the plain HTTP connection rule was triggered and block action was taken.") + + +class AbuseShieldMultipleRulesTest: + """Verify abuse_shield plugin can handle multiple rules with different thresholds. + + This test verifies that when multiple rules are configured, the first matching + rule triggers blocking. Combined rules (AND logic) with connection rate require + multiple physical connections and are better tested manually. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run multiple rules test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_multiple_rules() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_multi{AbuseShieldMultipleRulesTest._server_counter}' + AbuseShieldMultipleRulesTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with multiple rules with different thresholds.""" + name = f'ts_multi{AbuseShieldMultipleRulesTest._ts_counter}' + AbuseShieldMultipleRulesTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create plugin config with multiple rules - first match wins. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 60 + +rules: + - name: "lenient_limit" + filter: + max_req_rate: 100 + action: [log] + + - name: "strict_limit" + filter: + max_req_rate: 15 + action: [log, block] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 2 rules", + "Verify the abuse_shield plugin loaded with multiple rules.") + + def _test_multiple_rules(self) -> None: + """Verify a lenient first rule does not inherit a strict rule's debt.""" + tr = Test.AddTestRun("Trigger strict rule placed after a lenient rule") + + # Send requests via H2 rate client to exceed the strict_limit threshold. + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host localhost --port {self._ts.Variables.ssl_port} ' + f'--num-requests 50 --rate 100 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + self._ts.Disk.diags_log.Content += Testers.ExcludesExpression( + r'Rule "lenient_limit" matched', "Verify the lenient first rule did not inherit the strict rule's token state.") + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "strict_limit" matched for IP=.*actions=\[log,block\]', + "Verify the strict_limit rule was triggered and block action was taken.") + + +class AbuseShieldBlockExpirationTest: + """Verify abuse_shield plugin block expiration works correctly. + + After a block expires (duration_seconds), the IP should be able to + make requests again. + """ + + _server_counter: int = 0 + _ts_counter: int = 0 + + def __init__(self): + """Set up the test environment and run block expiration test scenarios.""" + self._setup_origin_server() + self._setup_ts() + self._test_block_expiration() + + def _setup_origin_server(self) -> None: + """Configure a simple HTTP/1.1 origin server.""" + name = f'origin_expire{AbuseShieldBlockExpirationTest._server_counter}' + AbuseShieldBlockExpirationTest._server_counter += 1 + + self._origin = Test.MakeOriginServer(name) + + self._origin.addResponse( + "sessionlog.json", { + "headers": "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n", + "timestamp": "1469733493.993", + "body": "" + }, { + "headers": + "HTTP/1.1 200 OK\r\nServer: origin\r\nCache-Control: max-age=300\r\nConnection: close\r\nContent-Length: 2\r\n\r\n", + "timestamp": "1469733493.993", + "body": "OK" + }) + + def _setup_ts(self) -> None: + """Configure ATS with a short block duration for testing expiration.""" + name = f'ts_expire{AbuseShieldBlockExpirationTest._ts_counter}' + AbuseShieldBlockExpirationTest._ts_counter += 1 + + self._ts = Test.MakeATSProcess(name, enable_tls=True, enable_cache=True) + self._ts.addDefaultSSLFiles() + + self._ts.Disk.ssl_multicert_yaml.AddLines( + ''' +ssl_multicert: +- dest_ip: '*' + ssl_cert_name: server.pem + ssl_key_name: server.key +'''.split('\n')) + self._ts.Disk.remap_config.AddLine(f'map / http://127.0.0.1:{self._origin.Variables.Port}/') + + self._ts.Disk.records_config.update( + { + 'proxy.config.diags.debug.enabled': 1, + 'proxy.config.diags.debug.tags': 'abuse_shield', + 'proxy.config.ssl.server.cert.path': self._ts.Variables.SSLDir, + 'proxy.config.ssl.server.private_key.path': self._ts.Variables.SSLDir, + }) + + # Create plugin config with a very short block duration (5 seconds) for testing. + self._ts.Disk.File(self._ts.Variables.CONFIGDIR + "/abuse_shield.yaml", id="abuse_shield_yaml", typename="ats:config") + self._ts.Disk.abuse_shield_yaml.AddLines( + ''' +global: + ip_tracking: + slots: 1000 + + blocking: + duration_seconds: 5 + +rules: + - name: "short_block" + filter: + max_req_rate: 10 + action: [log, block] + +enabled: true +'''.strip().split('\n')) + + self._ts.Disk.plugin_config.AddLine('abuse_shield.so abuse_shield.yaml') + + self._ts.Disk.diags_log.Content = Testers.ContainsExpression( + r"abuse_shield.*Plugin initialized with 1000 slots per tracker, 1 rules", + "Verify the abuse_shield plugin loaded with short block duration.") + + def _test_block_expiration(self) -> None: + """Trigger a block, wait for expiration, and verify requests work again.""" + # Step 1: Trigger a block by exceeding rate limit. + tr = Test.AddTestRun("Trigger block by exceeding rate limit") + + client_cmd = ( + f'{sys.executable} {Test.TestDirectory}/h2_rate_client.py ' + f'--host 127.0.0.1 --port {self._ts.Variables.ssl_port} ' + f'--num-requests 50 --rate 100 --path /') + tr.Processes.Default.Command = client_cmd + tr.Processes.Default.ReturnCode = 0 + tr.Processes.Default.StartBefore(self._origin) + tr.Processes.Default.StartBefore(self._ts) + tr.StillRunningAfter = self._ts + + # Verify blocking occurred. + self._ts.Disk.diags_log.Content += Testers.ContainsExpression( + r'Rule "short_block" matched for IP=.*actions=\[log,block\]', + "Verify the short_block rule was triggered and block action was taken.") + + # Step 2: Verify the block action and a subsequent rejection are both + # observable by operators. + tr = Test.AddTestRun("Verify the temporary block is active") + tr.Processes.Default.Command = (f'curl -k -s -o /dev/null --max-time 2 https://127.0.0.1:{self._ts.Variables.ssl_port}/') + tr.Processes.Default.ReturnCode = Any(28, 35, 52, 55, 56) + tr.StillRunningAfter = self._ts Review Comment: The first step starts the origin server, but only ATS is listed in `StillRunningAfter`. Because this test then runs multiple subsequent TestRuns without restarting the origin, AuTest may stop the origin between steps (processes not in `StillRunningAfter` can be torn down after the run), which can make later curl steps depend on cache state and become flaky. Keep the origin in `StillRunningAfter` at least through the follow-up validation steps (see tests/gold_tests/h2/h2enable.test.py:66-95 for the common pattern). ########## plugins/experimental/abuse_shield/abuse_shield.cc: ########## @@ -0,0 +1,1278 @@ +/** @file + + Abuse Shield Plugin - HTTP/2 error tracking and IP-based abuse detection. + + Uses the Udi "King of the Hill" algorithm for efficient, bounded-memory IP tracking. + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one or more contributor license + agreements. See the NOTICE file distributed with this work for additional information regarding + copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with the License. You may obtain + a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software distributed under the License + is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + or implied. See the License for the specific language governing permissions and limitations under + the License. +*/ + +#include <algorithm> +#include <chrono> +#include <cinttypes> +#include <cstring> +#include <ctime> +#include <exception> +#include <iomanip> +#include <memory> +#include <mutex> +#include <shared_mutex> +#include <sstream> +#include <string> +#include <type_traits> +#include <unordered_map> + +#include <sys/socket.h> + +#include "ts/ts.h" +#include "swoc/IPRange.h" +#include "swoc/BufferWriter.h" +#include "swoc/bwf_ip.h" + +#include "config.h" +#include "fingerprint.h" +#include "fingerprint_registry.h" +#include "ip_data.h" +#include "stats.h" +#include "logging.h" + +namespace +{ +using abuse_shield::dbg_ctl; +using abuse_shield::PLUGIN_NAME; + +// Optional log file for LOG action output. +TSTextLogObject g_log_object = nullptr; + +// Global action stats. +abuse_shield::ActionStats g_action_stats; + +// Named JAx VConn user-arg selected by global.fingerprint_registry. +int g_fingerprint_registry_index = -1; + +// Per-tracker stats. +abuse_shield::TrackerStats g_txn_stats; +abuse_shield::TrackerStats g_conn_stats; +abuse_shield::TrackerStats g_h2_stats; + +// Helper to convert IPAddr to string. +std::string +ip_to_string(const swoc::IPAddr &ip) +{ + swoc::LocalBufferWriter<64> writer; + writer.print("{}", ip); + return std::string(writer.view()); +} + +std::string +format_local_time(std::time_t time, const char *format) +{ + struct tm time_parts; + + if (localtime_r(&time, &time_parts) == nullptr) { + return "-"; + } + + std::ostringstream oss; + oss << std::put_time(&time_parts, format); + return oss.str(); +} + +// Helper to get current wall clock time as string. +std::string +current_time_str() +{ + auto now = std::chrono::system_clock::now(); + auto now_t = std::chrono::system_clock::to_time_t(now); + return format_local_time(now_t, "%Y-%m-%dT%H:%M:%S"); +} + +// ============================================================================ +// Global state +// ============================================================================ + +// Separate UDI tables for different event types, each with its own data type. +std::unique_ptr<abuse_shield::TxnTable> g_txn_tracker; ///< Transaction/request rate tracking +std::unique_ptr<abuse_shield::ConnTable> g_conn_tracker; ///< Connection rate tracking +std::unique_ptr<abuse_shield::H2Table> g_h2_tracker; ///< HTTP/2 error tracking + +/** Bounded block state which never evicts an unexpired block. */ +class BlockedIpTable +{ +public: + explicit BlockedIpTable(size_t capacity) : capacity_(capacity) { blocked_.reserve(capacity); } + + bool + block(const swoc::IPAddr &ip, uint64_t until_ms) + { + std::lock_guard lock(mutex_); + auto spot = blocked_.find(ip); + if (spot != blocked_.end()) { + spot->second = std::max(spot->second, until_ms); + return true; + } + + if (blocked_.size() >= capacity_) { + uint64_t now = abuse_shield::now_ms(); + std::erase_if(blocked_, [now](auto const &item) { return item.second <= now; }); + } + if (blocked_.size() >= capacity_) { + return false; + } + + blocked_.emplace(ip, until_ms); + return true; + } + + bool + is_blocked(const swoc::IPAddr &ip) + { + std::lock_guard lock(mutex_); + auto spot = blocked_.find(ip); + if (spot == blocked_.end()) { + return false; + } + if (spot->second <= abuse_shield::now_ms()) { + blocked_.erase(spot); + return false; + } + return true; + } + +private: + size_t capacity_; + std::mutex mutex_; + std::unordered_map<swoc::IPAddr, uint64_t> blocked_; +}; + +std::unique_ptr<BlockedIpTable> g_blocked_ips; + +std::shared_ptr<abuse_shield::Config> g_config; +std::shared_mutex g_config_mutex; // Protects g_config pointer swaps + +// Sync the metrics from a single Udi table to its stats. +void +sync_tracker_stats(abuse_shield::TxnTable *tracker, abuse_shield::TrackerStats &stats) +{ + if (tracker) { + TSStatIntSet(stats.slots_used, static_cast<int64_t>(tracker->slots_used())); + TSStatIntSet(stats.contests, static_cast<int64_t>(tracker->contests())); + TSStatIntSet(stats.contests_won, static_cast<int64_t>(tracker->contests_won())); + TSStatIntSet(stats.evictions, static_cast<int64_t>(tracker->evictions())); + } +} + +void +sync_tracker_stats(abuse_shield::ConnTable *tracker, abuse_shield::TrackerStats &stats) +{ + if (tracker) { + TSStatIntSet(stats.slots_used, static_cast<int64_t>(tracker->slots_used())); + TSStatIntSet(stats.contests, static_cast<int64_t>(tracker->contests())); + TSStatIntSet(stats.contests_won, static_cast<int64_t>(tracker->contests_won())); + TSStatIntSet(stats.evictions, static_cast<int64_t>(tracker->evictions())); + } +} + +void +sync_tracker_stats(abuse_shield::H2Table *tracker, abuse_shield::TrackerStats &stats) +{ + if (tracker) { + TSStatIntSet(stats.slots_used, static_cast<int64_t>(tracker->slots_used())); + TSStatIntSet(stats.contests, static_cast<int64_t>(tracker->contests())); + TSStatIntSet(stats.contests_won, static_cast<int64_t>(tracker->contests_won())); + TSStatIntSet(stats.evictions, static_cast<int64_t>(tracker->evictions())); + } +} + +// Sync the metrics from all Udi tables. +void +sync_all_tracker_stats() +{ + sync_tracker_stats(g_txn_tracker.get(), g_txn_stats); + sync_tracker_stats(g_conn_tracker.get(), g_conn_stats); + sync_tracker_stats(g_h2_tracker.get(), g_h2_stats); +} + +template <typename Table> +void +reset_tracker_stats(Table *tracker, abuse_shield::TrackerStats &stats) +{ + if (tracker) { + tracker->reset_metrics(); + } + TSStatIntSet(stats.events, 0); + TSStatIntSet(stats.events_untracked, 0); + TSStatIntSet(stats.scan_exhausted, 0); + sync_tracker_stats(tracker, stats); +} + +// ============================================================================ +// Rule evaluation +// ============================================================================ + +int +metric_rate(const abuse_shield::RuleFilter &filter, abuse_shield::RateMetric metric) +{ + switch (metric) { + case abuse_shield::RateMetric::REQUEST: + return filter.max_req_rate; + case abuse_shield::RateMetric::CONNECTION: + return filter.max_conn_rate; + case abuse_shield::RateMetric::H2_ERROR: + return filter.max_h2_error_rate; + } + return 0; +} + +double +metric_burst_multiplier(const abuse_shield::RuleFilter &filter, abuse_shield::RateMetric metric) +{ + switch (metric) { + case abuse_shield::RateMetric::REQUEST: + return filter.req_burst_multiplier; + case abuse_shield::RateMetric::CONNECTION: + return filter.conn_burst_multiplier; + case abuse_shield::RateMetric::H2_ERROR: + return filter.h2_burst_multiplier; + } + return 1.0; +} + +template <typename Table> +bool +rate_exceeded(Table *tracker, const abuse_shield::Rule &rule, const swoc::IPAddr &ip) +{ + auto slot = tracker ? tracker->find(ip) : nullptr; + return slot && slot->buckets.exceeded(rule.name); +} + +template <typename Table> +void +consume_rule_buckets(Table *tracker, const swoc::IPAddr &ip, const abuse_shield::Config &config, abuse_shield::RateMetric metric, + abuse_shield::TrackerStats &stats, uint64_t error_code = 0) +{ + typename Table::data_ptr slot; + bool consumed = false; + + for (const auto &rule : config.rules()) { + int rate = metric_rate(rule.filter, metric); + if (rate <= 0 || !config.rule_applies_to_ip(rule, ip)) { + continue; + } + + if (!slot) { + typename Table::ProcessStatus status; + slot = tracker->process_event(ip, 1, &status); + if (!slot) { + TSStatIntIncrement(stats.events_untracked, 1); + if (status == Table::ProcessStatus::NO_CANDIDATE) { + TSStatIntIncrement(stats.scan_exhausted, 1); + } + return; + } + } + + int burst = static_cast<int>(static_cast<double>(rate) * metric_burst_multiplier(rule.filter, metric)); + if constexpr (std::is_same_v<typename Table::data_type, abuse_shield::H2Data>) { + slot->consume(rule.name, rate, burst, error_code); + } else { + slot->consume(rule.name, rate, burst); + } + consumed = true; + } + + if (consumed) { + TSStatIntIncrement(stats.events, 1); + } +} + +/** Check if a rule's filter criteria match for the given IP. + * + * A rule matches only if ALL enabled filter criteria are satisfied (AND logic). + * Each filter uses token bucket rate limiting - a rate is exceeded when tokens < 0. + * Fingerprint values and methods use OR logic with one another. The resulting + * fingerprint criterion is ANDed with any enabled rate criteria. + * + * @param[in] rule The rule containing filter criteria to check. + * @param[in] ip The IP address to evaluate. + * @return True if all enabled filter criteria are satisfied, false otherwise. + */ +bool +rule_matches(const abuse_shield::Rule &rule, const swoc::IPAddr &ip, const abuse_shield::Config &config, + const abuse_shield::FingerprintResults *fingerprints, std::string_view &matched_method, + std::string_view &matched_fingerprint) +{ + const auto &f = rule.filter; + + if (!config.rule_applies_to_ip(rule, ip)) { + return false; + } + + if (f.max_req_rate == 0 && f.max_conn_rate == 0 && f.max_h2_error_rate == 0 && !f.has_fingerprints()) { + return false; + } + + if (f.has_fingerprints()) { + if (!fingerprints) { + return false; + } + + bool fingerprint_matched = false; + for (const auto &[method, configured_values] : f.fingerprints) { + auto computed = fingerprints->find(method); + if (computed != fingerprints->end() && configured_values.contains(computed->second)) { + matched_method = computed->first; + matched_fingerprint = computed->second; + fingerprint_matched = true; + break; + } + } + if (!fingerprint_matched) { + return false; + } + } + + if (f.max_req_rate > 0 && !rate_exceeded(g_txn_tracker.get(), rule, ip)) { + return false; + } + if (f.max_conn_rate > 0 && !rate_exceeded(g_conn_tracker.get(), rule, ip)) { + return false; + } + if (f.max_h2_error_rate > 0 && !rate_exceeded(g_h2_tracker.get(), rule, ip)) { + return false; + } + + return true; // All enabled filters matched +} + +abuse_shield::RuleMatch +evaluate_rate_rules(const swoc::IPAddr &ip, const abuse_shield::Config &config) +{ + for (const auto &rule : config.rules()) { + if (rule.filter.has_fingerprints()) { + continue; + } + + std::string_view matched_method; + std::string_view matched_fingerprint; + if (rule_matches(rule, ip, config, nullptr, matched_method, matched_fingerprint)) { + Dbg(dbg_ctl, "Rule matched: %s", rule.name.c_str()); + return abuse_shield::RuleMatch{&rule, rule.actions, {}, {}}; + } + } + return abuse_shield::RuleMatch{}; +} + +abuse_shield::RuleMatch +evaluate_fingerprint_rules(const swoc::IPAddr &ip, const abuse_shield::Config &config, + const abuse_shield::FingerprintResults &fingerprints) +{ + for (const auto &rule : config.rules()) { + if (!rule.filter.has_fingerprints()) { + continue; + } + + std::string_view matched_method; + std::string_view matched_fingerprint; + if (rule_matches(rule, ip, config, &fingerprints, matched_method, matched_fingerprint)) { + Dbg(dbg_ctl, "Fingerprint rule matched: %s", rule.name.c_str()); + return abuse_shield::RuleMatch{&rule, rule.actions, matched_method, matched_fingerprint}; + } + } + return abuse_shield::RuleMatch{}; +} + +// ============================================================================ +// Action execution +// ============================================================================ + +/** Store block state independently from the evictable rate tables. */ +bool +block_ip(const swoc::IPAddr &ip, uint64_t until_ms) +{ + return g_blocked_ips && g_blocked_ips->block(ip, until_ms); +} + +/** Execute the log action while respecting the log rate limit. + * + * @param[in] match The rule match result. + * @param[in] ip The client IP address. + * @param[in] config The current configuration. + */ +void +execute_log_action(const abuse_shield::RuleMatch &match, const swoc::IPAddr &ip, const abuse_shield::Config &config) +{ + uint64_t log_interval_ms = static_cast<uint64_t>(config.log_interval_sec()) * 1000; + uint64_t now = abuse_shield::now_ms(); + + // Find the most recent log time across all trackers for this IP. + uint64_t most_recent_log = 0; + auto txn_slot = g_txn_tracker ? g_txn_tracker->find(ip) : nullptr; + auto conn_slot = g_conn_tracker ? g_conn_tracker->find(ip) : nullptr; + auto h2_slot = g_h2_tracker ? g_h2_tracker->find(ip) : nullptr; + + // Fingerprint-only rules do not otherwise need a rate-tracking slot. Use the + // bounded connection table to retain their per-IP log interval state. + if (!txn_slot && !conn_slot && !h2_slot && g_conn_tracker) { + conn_slot = g_conn_tracker->process_event(ip, 1); + } + + if (txn_slot) { + most_recent_log = std::max(most_recent_log, txn_slot->last_logged.load(std::memory_order_relaxed)); + } + if (conn_slot) { + most_recent_log = std::max(most_recent_log, conn_slot->last_logged.load(std::memory_order_relaxed)); + } + if (h2_slot) { + most_recent_log = std::max(most_recent_log, h2_slot->last_logged.load(std::memory_order_relaxed)); + } + + // Only log if enough time has passed since the last log for this IP. + if (!abuse_shield::log_interval_elapsed(now, most_recent_log, log_interval_ms)) { + return; + } + + // Try to claim the log opportunity atomically using the first available slot. + bool claimed = false; + + if (txn_slot) { + claimed = abuse_shield::claim_log_interval(txn_slot->last_logged, now, log_interval_ms); + } else if (conn_slot) { + claimed = abuse_shield::claim_log_interval(conn_slot->last_logged, now, log_interval_ms); + } else if (h2_slot) { + claimed = abuse_shield::claim_log_interval(h2_slot->last_logged, now, log_interval_ms); + } + + if (!claimed) { + return; + } + + // Update last_logged in all slots. + if (txn_slot) { + txn_slot->last_logged.store(now, std::memory_order_relaxed); + } + if (conn_slot) { + conn_slot->last_logged.store(now, std::memory_order_relaxed); + } + if (h2_slot) { + h2_slot->last_logged.store(now, std::memory_order_relaxed); + } + + TSStatIntIncrement(g_action_stats.actions_logged, 1); + + // Get token state for logging. + int32_t req_tokens = txn_slot ? txn_slot->buckets.tokens(match.rule->name) : 0; + int32_t conn_tokens = conn_slot ? conn_slot->buckets.tokens(match.rule->name) : 0; + int32_t h2_tokens = h2_slot ? h2_slot->buckets.tokens(match.rule->name) : 0; + + std::string fingerprint; + if (!match.fingerprint.empty()) { + fingerprint = " fingerprint="; + fingerprint.append(match.fingerprint_method); + fingerprint.push_back(':'); + fingerprint.append(match.fingerprint); + } + + if (g_log_object) { + TSTextLogObjectWrite(g_log_object, "Rule \"%s\" matched for IP=%s:%s actions=[%s] req_tokens=%d conn_tokens=%d h2_tokens=%d", + match.rule->name.c_str(), ip_to_string(ip).c_str(), fingerprint.c_str(), + abuse_shield::actions_to_string(match.actions).c_str(), req_tokens, conn_tokens, h2_tokens); + } else { + TSError("[%s] Rule \"%s\" matched for IP=%s:%s actions=[%s] req_tokens=%d conn_tokens=%d h2_tokens=%d", PLUGIN_NAME, + match.rule->name.c_str(), ip_to_string(ip).c_str(), fingerprint.c_str(), + abuse_shield::actions_to_string(match.actions).c_str(), req_tokens, conn_tokens, h2_tokens); + } Review Comment: The log format always prints `IP=%s:%s`, but `fingerprint` already includes a leading space and is empty for non-fingerprint rules. This yields awkward output like `IP=1.2.3.4:` (trailing colon) or `IP=1.2.3.4: fingerprint=...` (colon+space coming from two different places). Consider formatting the fingerprint as an optional suffix and remove the hard-coded colon so the log line is consistent and unambiguous. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
