maskit commented on code in PR #13586: URL: https://github.com/apache/trafficserver/pull/13586#discussion_r3896931207
########## plugins/experimental/abuse_shield/abuse_shield.cc: ########## @@ -0,0 +1,1295 @@ +/** @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 (now - most_recent_log < log_interval_ms) { + return; + } + + // Try to claim the log opportunity atomically using the first available slot. + bool claimed = false; + uint64_t expected = 0; + + if (txn_slot) { + expected = txn_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = txn_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } else if (conn_slot) { + expected = conn_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = conn_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } else if (h2_slot) { + expected = h2_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = h2_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } + + 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); + } +} + +enum class CloseHandling { + SOCKET_SHUTDOWN, + REENABLE_ERROR, +}; + +/** Execute actions for a matched rule. + * + * @param[in] match The rule match result. + * @param[in] ip The client IP address. + * @param[in] vconn The virtual connection (for close action). + * @param[in] config The current configuration. + */ +bool +execute_actions(const abuse_shield::RuleMatch &match, const swoc::IPAddr &ip, TSVConn vconn, const abuse_shield::Config &config, + CloseHandling close_handling = CloseHandling::SOCKET_SHUTDOWN) +{ + TSStatIntIncrement(g_action_stats.rules_matched, 1); + + if (abuse_shield::has_action(match.actions, abuse_shield::Action::BLOCK)) { + uint64_t block_until = abuse_shield::now_ms() + (config.block_duration_sec() * 1000); + if (block_ip(ip, block_until)) { + TSStatIntIncrement(g_action_stats.actions_blocked, 1); + Dbg(dbg_ctl, "Blocking IP %s for %d seconds (rule: %s)", ip_to_string(ip).c_str(), config.block_duration_sec(), + match.rule->name.c_str()); + } else { + TSStatIntIncrement(g_action_stats.actions_block_failed, 1); + TSError("[%s] Block table is full; could not block %s for rule '%s'", PLUGIN_NAME, ip_to_string(ip).c_str(), + match.rule->name.c_str()); + } + } + + bool should_close = abuse_shield::has_action(match.actions, abuse_shield::Action::CLOSE); + if (should_close) { + if (close_handling == CloseHandling::SOCKET_SHUTDOWN) { + int fd = TSVConnFdGet(vconn); + if (fd >= 0 && shutdown(fd, SHUT_RDWR) == 0) { + TSStatIntIncrement(g_action_stats.actions_closed, 1); + Dbg(dbg_ctl, "Closing connection from %s (rule: %s)", ip_to_string(ip).c_str(), match.rule->name.c_str()); + } else { + TSStatIntIncrement(g_action_stats.actions_close_failed, 1); + TSError("[%s] Could not close the connection from %s for rule '%s'", PLUGIN_NAME, ip_to_string(ip).c_str(), + match.rule->name.c_str()); + } + } else { + // The hook is rejected with TSVConnReenableEx by the caller. + TSStatIntIncrement(g_action_stats.actions_closed, 1); + } + } + + // Log if logging is configured (independent of block/close actions). + if (abuse_shield::has_action(match.actions, abuse_shield::Action::LOG)) { + execute_log_action(match, ip, config); + } + + return should_close; +} + +// ============================================================================ +// Hook handlers +// ============================================================================ + +// Helper struct for error info. +struct H2Errors { + uint32_t cls{0}; ///< Error class (1 = connection, 2 = stream) + uint64_t code{0}; ///< HTTP/2 error code +}; + +/** Process HTTP/2 response errors. + * + * Tracks HTTP/2 stream and connection errors using token bucket rate limiting. + * + * @param[in] txnp The transaction being closed. + * @param[in] vconn The virtual connection. + * @param[in] ip The client IP address. + * @param[in] config The current configuration. + */ +void +process_h2_response(TSHttpTxn txnp, TSVConn vconn, const swoc::IPAddr &ip, const abuse_shield::Config &config) +{ + if (!g_h2_tracker) { + return; + } + + // Get HTTP/2 errors. + H2Errors received_error; // Error received from the client. + H2Errors sent_error; // Error sent to the client. + TSHttpTxnClientReceivedErrorGet(txnp, &received_error.cls, &received_error.code); Review Comment: There are 4 combinations (direction x error-class), but only 2 of them are checked. We may want to track all of them for attack like MadeYouReset. ########## plugins/experimental/abuse_shield/abuse_shield.cc: ########## @@ -0,0 +1,1295 @@ +/** @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 (now - most_recent_log < log_interval_ms) { + return; + } + + // Try to claim the log opportunity atomically using the first available slot. + bool claimed = false; + uint64_t expected = 0; + + if (txn_slot) { + expected = txn_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = txn_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } else if (conn_slot) { + expected = conn_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = conn_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } else if (h2_slot) { + expected = h2_slot->last_logged.load(std::memory_order_relaxed); + if (now - expected >= log_interval_ms) { + claimed = h2_slot->last_logged.compare_exchange_weak(expected, now, std::memory_order_relaxed); + } + } + + 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); + } +} + +enum class CloseHandling { + SOCKET_SHUTDOWN, + REENABLE_ERROR, +}; + +/** Execute actions for a matched rule. + * + * @param[in] match The rule match result. + * @param[in] ip The client IP address. + * @param[in] vconn The virtual connection (for close action). + * @param[in] config The current configuration. + */ +bool +execute_actions(const abuse_shield::RuleMatch &match, const swoc::IPAddr &ip, TSVConn vconn, const abuse_shield::Config &config, + CloseHandling close_handling = CloseHandling::SOCKET_SHUTDOWN) +{ + TSStatIntIncrement(g_action_stats.rules_matched, 1); + + if (abuse_shield::has_action(match.actions, abuse_shield::Action::BLOCK)) { + uint64_t block_until = abuse_shield::now_ms() + (config.block_duration_sec() * 1000); + if (block_ip(ip, block_until)) { + TSStatIntIncrement(g_action_stats.actions_blocked, 1); + Dbg(dbg_ctl, "Blocking IP %s for %d seconds (rule: %s)", ip_to_string(ip).c_str(), config.block_duration_sec(), + match.rule->name.c_str()); + } else { + TSStatIntIncrement(g_action_stats.actions_block_failed, 1); + TSError("[%s] Block table is full; could not block %s for rule '%s'", PLUGIN_NAME, ip_to_string(ip).c_str(), + match.rule->name.c_str()); + } + } + + bool should_close = abuse_shield::has_action(match.actions, abuse_shield::Action::CLOSE); + if (should_close) { + if (close_handling == CloseHandling::SOCKET_SHUTDOWN) { + int fd = TSVConnFdGet(vconn); + if (fd >= 0 && shutdown(fd, SHUT_RDWR) == 0) { + TSStatIntIncrement(g_action_stats.actions_closed, 1); + Dbg(dbg_ctl, "Closing connection from %s (rule: %s)", ip_to_string(ip).c_str(), match.rule->name.c_str()); + } else { + TSStatIntIncrement(g_action_stats.actions_close_failed, 1); + TSError("[%s] Could not close the connection from %s for rule '%s'", PLUGIN_NAME, ip_to_string(ip).c_str(), + match.rule->name.c_str()); + } + } else { + // The hook is rejected with TSVConnReenableEx by the caller. + TSStatIntIncrement(g_action_stats.actions_closed, 1); + } + } + + // Log if logging is configured (independent of block/close actions). + if (abuse_shield::has_action(match.actions, abuse_shield::Action::LOG)) { + execute_log_action(match, ip, config); + } + + return should_close; +} + +// ============================================================================ +// Hook handlers +// ============================================================================ + +// Helper struct for error info. +struct H2Errors { + uint32_t cls{0}; ///< Error class (1 = connection, 2 = stream) + uint64_t code{0}; ///< HTTP/2 error code +}; + +/** Process HTTP/2 response errors. + * + * Tracks HTTP/2 stream and connection errors using token bucket rate limiting. + * + * @param[in] txnp The transaction being closed. + * @param[in] vconn The virtual connection. + * @param[in] ip The client IP address. + * @param[in] config The current configuration. + */ +void +process_h2_response(TSHttpTxn txnp, TSVConn vconn, const swoc::IPAddr &ip, const abuse_shield::Config &config) +{ + if (!g_h2_tracker) { + return; + } + + // Get HTTP/2 errors. + H2Errors received_error; // Error received from the client. + H2Errors sent_error; // Error sent to the client. + TSHttpTxnClientReceivedErrorGet(txnp, &received_error.cls, &received_error.code); Review Comment: There are 4 combinations (direction x error-class), but only 2 of them are checked. We may want to track all of them for attacks like MadeYouReset. -- 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]
