AID-I-003.002: Identity, Session & API Rate-Limit Throttling

Version 1.20261015 · Actionable sub-technique

Apply atomic admission-control throttles at the API gateway or application edge for authenticated principals, tenants, sessions, API keys, and low-trust or unauthenticated callers. For AI endpoints exposed to untrusted clients, consume trusted bot or abuse signals to apply additional bounded restrictions to suspicious traffic, including correlated rotating sources where the deployed service supports them. Preserve the base identity allowance when adaptive restrictions change. This sub-technique owns identity binding, base trust-tier limits, server-derived admission costs, quota enforcement and bounded adaptive responses; it does not require a custom ML detector.

Pillars: infra, app

Lifecycle: response, operation

Scope boundary

Owns request-admission throttling by principal, tenant, session, API key or trust tier, and additional restrictions on a bounded suspicious traffic set, at the gateway or application edge. It consumes trusted bot or abuse findings without owning their detection or attribution. It does not enforce cumulative GPU, CPU, token, tool-call, retry, concurrency or spend budgets and does not decide the disposition of one high-risk action.

AID-I-003.003: AID-I-003.002 limits request admission rates by identity or session; AID-I-003.003 limits cumulative resource, token, concurrency, retry, tool-call, and spend consumption. Expensive low-frequency abuse can pass a request-rate limit but exceed a resource budget, while a high request rate may be throttled before cumulative compute becomes abnormal.

AID-I-003.004: AID-I-003.002 throttles a stream of API requests; AID-I-003.004 blocks, defers, reroutes, or escalates one proposed high-risk agent action before side effects. A per-identity rate limit does not determine whether an individual action is safe to execute.

Implementation guidance

AID-I-003.002-G001

Enforce atomic identity-bound request admission limits with server-derived costs.
Enforce the base admission allowance

Run this gate before inference or tool dispatch. Derive tenant/principal identity from authenticated transport and request cost from server-side token/tool/complexity accounting. Resolve identity at the trusted gateway; caller-selected IP, User-Agent, cookie, API-key header text, or forwarded headers are not authenticated principals. Bind a verified API key or session to its server-side principal rather than granting a fresh allowance for each header or network change. Verify a signed, monotonic policy before constructing the limiter. Use integer cost units, one durable idempotency key per request, Redis server time, and one Lua transaction so concurrent callers cannot overspend. Storage failure is ERROR, not a normal 429 denial.

Keep the base allowance independent of short-lived bot or abuse scores. A separate adaptive restriction may deny or throttle requests, but cannot replace this gate or increase its allowance. The example namespaces state by policy_version; switching versions creates another bucket. Do not translate every risk-score or temporary restriction change into a new base policy version. Before a planned base-policy or entitlement transition, explicitly preserve or migrate consumed allowance and retry decisions; the example does not implement that migration. Removing an adaptive restriction resumes the existing base state, not a fresh full bucket. Cumulative token, concurrency, tool-call and spend reservations remain in AID-I-003.003.

Idempotent Redis token bucket
# File: gateway/atomic_rate_limit.py
from __future__ import annotations

import hashlib
import math
from dataclasses import dataclass
from typing import Callable

import redis
from redis.exceptions import RedisError


TOKEN_BUCKET_LUA = r"""
local bucket = KEYS[1]
local decision_key = KEYS[2]
local capacity_units = tonumber(ARGV[1])
local refill_units_per_second = tonumber(ARGV[2])
local cost_units = tonumber(ARGV[3])
local ttl_ms = tonumber(ARGV[4])
local max_safe_units = 9007199254740

local prior = redis.call("GET", decision_key)
if prior then
  local decoded = cjson.decode(prior)
  return {
    decoded.allowed, decoded.remaining_units,
    decoded.remaining_milliunits, decoded.decided_at_ms, 1
  }
end

if not capacity_units or capacity_units <= 0
   or capacity_units ~= math.floor(capacity_units) or capacity_units > max_safe_units
   or not refill_units_per_second or refill_units_per_second <= 0
   or refill_units_per_second ~= math.floor(refill_units_per_second)
   or refill_units_per_second > max_safe_units
   or not cost_units or cost_units <= 0 or cost_units ~= math.floor(cost_units)
   or cost_units > capacity_units
   or not ttl_ms or ttl_ms <= 0 or ttl_ms ~= math.floor(ttl_ms) then
  return redis.error_reply("invalid token-bucket input")
end
local capacity_milliunits = capacity_units * 1000
local cost_milliunits = cost_units * 1000

local clock = redis.call("TIME")
local now_ms = clock[1] * 1000 + math.floor(clock[2] / 1000)
local values = redis.call("HMGET", bucket, "balance_milliunits", "last_ms")
local balance = tonumber(values[1])
local last_ms = tonumber(values[2])
if (values[1] and not balance) or (values[2] and not last_ms) then
  return redis.error_reply("malformed token-bucket state")
end
if not balance then
  balance = capacity_milliunits
  last_ms = now_ms
end
if last_ms > now_ms or balance < 0 or balance > capacity_milliunits
   or balance ~= math.floor(balance) then
  return redis.error_reply("invalid token-bucket state")
end

local elapsed_ms = now_ms - last_ms
local full_refill_ms = math.ceil(capacity_milliunits / refill_units_per_second)
if elapsed_ms >= full_refill_ms then
  balance = capacity_milliunits
else
  balance = math.min(
    capacity_milliunits,
    balance + elapsed_ms * refill_units_per_second
  )
end
local allowed = 0
if cost_milliunits <= balance then
  balance = balance - cost_milliunits
  allowed = 1
end
local remaining_units = math.floor(balance / 1000)
redis.call("HSET", bucket, "balance_milliunits", balance, "last_ms", now_ms)
redis.call("PEXPIRE", bucket, ttl_ms)
local encoded = cjson.encode({
  allowed = allowed,
  remaining_units = remaining_units,
  remaining_milliunits = balance,
  decided_at_ms = now_ms
})
redis.call("SET", decision_key, encoded, "PX", ttl_ms, "NX")
return {allowed, remaining_units, balance, now_ms, 0}
"""


@dataclass(frozen=True)
class VerifiedRatePolicy:
    policy_sha256: str
    policy_version: int
    capacity_units: int
    refill_units_per_second: int
    state_ttl_ms: int


@dataclass(frozen=True)
class RateDecision:
    outcome: str
    allowed: bool
    remaining_units: int
    remaining_milliunits: int
    decided_at_ms: int
    replayed: bool
    evidence: dict[str, str]


@dataclass(frozen=True)
class EvidenceWriter:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict[str, object]], dict[str, str]]


def digest_key(value: str) -> str:
    if not value:
        raise ValueError("empty identity or request key")
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def consume(
    *,
    client: redis.Redis,
    tenant_id: str,
    principal_id: str,
    request_id: str,
    cost_units: int,
    policy: VerifiedRatePolicy,
    evidence: EvidenceWriter,
) -> RateDecision:
    integers = (
        policy.policy_version, policy.capacity_units,
        policy.refill_units_per_second, policy.state_ttl_ms, cost_units,
    )
    if any(isinstance(value, bool) or not isinstance(value, int) or value <= 0 for value in integers):
        raise ValueError("rate policy and cost require positive integers")
    if any(value > 9_007_199_254_740 for value in integers):
        raise ValueError("rate values exceed exact Redis Lua integer range")
    if cost_units > policy.capacity_units:
        raise ValueError("request cost exceeds bucket capacity")
    full_refill_ms = math.ceil(
        policy.capacity_units * 1000 / policy.refill_units_per_second
    )
    if policy.state_ttl_ms < full_refill_ms:
        raise ValueError("state TTL must cover a full bucket refill")
    scope_hash = digest_key(tenant_id + "\x00" + principal_id)
    request_hash = digest_key(request_id)
    # The shared hash tag keeps both Lua keys in one Redis Cluster slot.
    slot = f"{policy.policy_version}:{scope_hash}"
    bucket_key = f"rate:v3:{{{slot}}}:bucket"
    decision_key = f"rate:v3:{{{slot}}}:decision:{request_hash}"
    script = client.register_script(TOKEN_BUCKET_LUA)
    try:
        allowed, remaining, remaining_milliunits, decided_at, replayed = script(
            keys=[bucket_key, decision_key],
            args=[
                policy.capacity_units,
                policy.refill_units_per_second,
                cost_units,
                policy.state_ttl_ms,
            ],
        )
    except RedisError as exc:
        evidence.append_and_verify({
            "schema_version": "aidefend.rate-decision.v3",
            "control": "AID-I-003.002",
            "request_id_sha256": request_hash,
            "scope_id_sha256": scope_hash,
            "policy_sha256": policy.policy_sha256,
            "error_type": type(exc).__name__,
            "outcome": "ERROR",
        })
        raise RuntimeError("rate-limit authority unavailable") from exc

    permitted = int(allowed) == 1
    receipt = evidence.append_and_verify({
        "schema_version": "aidefend.rate-decision.v3",
        "control": "AID-I-003.002",
        "request_id_sha256": request_hash,
        "scope_id_sha256": scope_hash,
        "policy_sha256": policy.policy_sha256,
        "policy_version": policy.policy_version,
        "cost_units": cost_units,
        "remaining_units": int(remaining),
        "remaining_milliunits": int(remaining_milliunits),
        "decided_at_ms": int(decided_at),
        "replayed": int(replayed) == 1,
        "outcome": "PASS" if permitted else "FAIL",
    })
    return RateDecision(
        outcome="PASS" if permitted else "FAIL",
        allowed=permitted,
        remaining_units=int(remaining),
        remaining_milliunits=int(remaining_milliunits),
        decided_at_ms=int(decided_at),
        replayed=int(replayed) == 1,
        evidence=receipt,
    )
Verify complete request coverage

Deploy the gate immediately before every governed inference/tool dispatch, use one authoritative Redis/Valkey cluster per policy scope or a proven globally serialized service, and return 429 only for a verified limit denial. Retain policy envelope, identity and cost-calculator versions, Lua/script digest, request population root, decision/evidence receipts, Redis topology and durability settings, and concurrency test results. A separate replay enumerates gateway access logs and requires one decision for every request, replays idempotent duplicates, races more total cost than capacity, advances refill boundaries, sends sub-token-interval traffic continuously for at least one full refill window, and proves that milliunit credit is never discarded; it also injects Redis failover, timeouts, malformed state, and old rate:v2 keys.

Also replay source IP and User-Agent changes under the same verified principal, and adaptive restriction activation and removal. Confirm that these changes neither select a new base bucket nor restore consumed allowance. A permitted request proves admission under this quota only; it does not establish that the request is benign.

AID-I-003.002-G002

Apply stricter admission limits to anonymous, lower-trust, or suspicious traffic using trusted bot and abuse signals.
1. Select exposed AI endpoints and response policy

Use adaptive restrictions where public or otherwise untrusted clients can invoke AI chat, inference, retrieval, or generation-job submission endpoints. Keep the base identity allowance for internal and public deployments; a bot-management integration is an optional implementation path, not a requirement to build a new ML detector. The API/platform owner selects the protected host, route and method, normal traffic population, acceptable false-positive and latency limits, observation window, restriction duration and rollback conditions from the versioned deployment policy. Distinguish legitimate API clients, batch jobs and AI agents from prohibited automation: a bot score is not a verdict of malicious intent.

2. Resolve the base entitlement once

Resolve one authoritative base tier from authenticated identity and contract/workforce entitlements, then signature-verify the immutable rate-policy reference and call the preceding atomic gate once for every request reaching this admission stage. Do not rely on global and consumer plugin precedence or treat a plugin instance_name as a policy reference. Additional abuse restrictions may further deny or throttle traffic, but must not replace, reset or increase the base allowance. Use temporary restrictions for short-lived risk changes; do not feed each bot-score change into rate_policy_ref or policy_version. Base entitlement changes require the state-transition handling described in AID-I-003.002-G001.

The code's principal_id = "anonymous" is one shared anonymous bucket per tenant, not an individual visitor quota. Size that aggregate guardrail for the endpoint's anonymous workload. A trusted edge can apply additional visitor or suspicious-traffic restrictions; IP and TLS fingerprints must not be relabeled as authenticated principals. Unverifiable callers remain in the least-privileged tier, and failure of the base identity, entitlement, policy or quota authority must not grant a higher allowance.

# File: gateway/trust_tier_rate_gate.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable

import redis

from gateway.atomic_rate_limit import (
    EvidenceWriter as AtomicEvidenceWriter,
    RateDecision,
    VerifiedRatePolicy,
    consume as consume_atomic,
    digest_key,
)


@dataclass(frozen=True)
class AuthenticatedIdentity:
    tenant_id: str
    principal_id: str
    authentication_class: str
    authentication_receipt_sha256: str


@dataclass(frozen=True)
class VerifiedTierPolicy:
    policy_sha256: str
    policy_version: int
    tier_order: tuple[str, ...]
    rate_policy_ref_by_tier: dict[str, str]
    anonymous_tier: str


@dataclass(frozen=True)
class EntitlementAuthority:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    read_verified_tier: Callable[[AuthenticatedIdentity], dict[str, object] | None]


@dataclass(frozen=True)
class VerifiedRatePolicyEnvelope:
    rate_policy_ref: str
    policy: VerifiedRatePolicy
    authority_receipt_sha256: str


@dataclass(frozen=True)
class RatePolicyAuthority:
    """Resolve and signature-verify an exact immutable rate-policy reference."""
    resolve_verified: Callable[[str], VerifiedRatePolicyEnvelope]


@dataclass(frozen=True)
class AtomicRateGate:
    client: redis.Redis
    policies: RatePolicyAuthority
    evidence: AtomicEvidenceWriter

    def consume(
        self, *, tenant_id: str, principal_id: str, request_id: str,
        cost_units: int, rate_policy_ref: str,
    ) -> tuple[RateDecision, VerifiedRatePolicyEnvelope]:
        envelope = self.policies.resolve_verified(rate_policy_ref)
        if (
            envelope.rate_policy_ref != rate_policy_ref
            or not envelope.authority_receipt_sha256
            or envelope.policy.policy_sha256 == ""
        ):
            raise PermissionError("verified rate-policy envelope binding mismatch")
        decision = consume_atomic(
            client=self.client,
            tenant_id=tenant_id,
            principal_id=principal_id,
            request_id=request_id,
            cost_units=cost_units,
            policy=envelope.policy,
            evidence=self.evidence,
        )
        return decision, envelope


@dataclass(frozen=True)
class EvidenceWriter:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict[str, object]], dict[str, str]]


def select_tier_and_consume(
    *,
    identity: AuthenticatedIdentity | None,
    tenant_id: str,
    request_id: str,
    cost_units: int,
    policy: VerifiedTierPolicy,
    entitlements: EntitlementAuthority,
    rate_gate: AtomicRateGate,
    evidence: EvidenceWriter,
) -> dict[str, object]:
    if (
        not policy.tier_order
        or len(set(policy.tier_order)) != len(policy.tier_order)
        or policy.tier_order[0] != policy.anonymous_tier
        or policy.anonymous_tier not in policy.rate_policy_ref_by_tier
        or set(policy.tier_order) != set(policy.rate_policy_ref_by_tier)
        or any(not value for value in policy.rate_policy_ref_by_tier.values())
    ):
        raise ValueError("invalid signed tier policy")

    tier = policy.anonymous_tier
    entitlement_receipt = None
    principal_id = "anonymous"
    auth_receipt = None
    if identity is not None:
        if identity.tenant_id != tenant_id:
            raise PermissionError("identity tenant mismatch")
        principal_id = identity.principal_id
        auth_receipt = identity.authentication_receipt_sha256
        entitlement = entitlements.read_verified_tier(identity)
        if entitlement is not None:
            if (
                entitlement.get("tenant_id") != tenant_id
                or entitlement.get("principal_id") != principal_id
                or entitlement.get("tier") not in policy.rate_policy_ref_by_tier
                or entitlement.get("policy_sha256") != policy.policy_sha256
            ):
                raise PermissionError("tier entitlement binding mismatch")
            tier = str(entitlement["tier"])
            entitlement_receipt = entitlement["receipt_sha256"]

    rate_policy_ref = policy.rate_policy_ref_by_tier[tier]
    decision, rate_envelope = rate_gate.consume(
        tenant_id=tenant_id,
        principal_id=principal_id,
        request_id=request_id,
        cost_units=cost_units,
        rate_policy_ref=rate_policy_ref,
    )
    decision_receipt_sha256 = decision.evidence.get("receipt_sha256")
    if (
        rate_envelope.rate_policy_ref != rate_policy_ref
        or decision.outcome not in {"PASS", "FAIL"}
        or decision.allowed != (decision.outcome == "PASS")
        or not isinstance(decision_receipt_sha256, str)
        or len(decision_receipt_sha256) != 64
    ):
        raise PermissionError("atomic gate decision binding mismatch")
    receipt = evidence.append_and_verify({
        "schema_version": "aidefend.trust-tier-rate.v1",
        "control": "AID-I-003.002",
        "tenant_id_sha256": digest_key(tenant_id),
        "principal_id_sha256": digest_key(principal_id),
        "request_id_sha256": digest_key(request_id),
        "tier": tier,
        "tier_policy_sha256": policy.policy_sha256,
        "tier_policy_version": policy.policy_version,
        "rate_policy_ref": rate_policy_ref,
        "rate_policy_sha256": rate_envelope.policy.policy_sha256,
        "rate_policy_version": rate_envelope.policy.policy_version,
        "rate_policy_authority_receipt_sha256": rate_envelope.authority_receipt_sha256,
        "authentication_receipt_sha256": auth_receipt,
        "entitlement_receipt_sha256": entitlement_receipt,
        "rate_decision_receipt_sha256": decision_receipt_sha256,
        "outcome": decision.outcome,
    })
    return {"allowed": decision.allowed, "outcome": decision.outcome, "evidence": receipt}

Verify that every lower base tier is no more permissive in admission capacity and refill rate than the tier above it. Keep resource, concurrency and spend accounting in AID-I-003.003. Requests rejected at the edge never dispatch work; requests admitted by the edge still pass the base quota and applicable resource-budget gates. An edge allow, verified-bot classification or partner exception cannot bypass these gates.

3. Consume trusted bot and abuse signals

Prefer the existing bot-management or API-protection service's native behavioral scoring, detections and available activity correlation. Candidate inputs include source IP/prefix and ASN reputation, HTTP/TLS fingerprints, User-Agent consistency, request cadence, endpoint sequence, retries and authenticated account/session activity. Relate only the signals the chosen service actually exposes; do not invent a universal botnet identifier or claim that a score alone links rotating accounts. Existing AID-D-005.007 findings can supply AI-specific retry, fan-out, token or cost-abuse signals. Detection stays with that producer; this method applies a bounded response.

For Akamai Bot Manager, configure the protected resource and its supported detection/response policy, use the deployment's native score segments and review traffic in Web Security Analytics. For Cloudflare, use Bot Management fields with Advanced Rate Limiting on the protected zone. Confirm edition, API/mobile versus browser telemetry, score direction, missing-value behavior and supported actions; provider scores are not interchangeable. Prefer enforcement inside the trusted edge. If signals must reach the gateway, strip client-supplied copies, accept replacements only over the authenticated edge-to-origin path, bind them to the request or subject and observation time, and reject stale or mismatched values. Restrict origin ingress to that path and test direct-origin bypass.

4. Aggregate the suspicious traffic without fragmenting counters

Separate matching criteria from counting characteristics. Matching selects a bounded suspicious traffic set; counting decides which matching requests share a limit. Putting IP, User-Agent, ASN and fingerprint together into a key creates a separate allowance for every distinct combination. When the product provides a validated activity group, use its supported grouping path. Otherwise scope a rule to a reviewed combination of signals and the exact AI endpoint. A common ASN, adjacent IP addresses or a common JA4 alone is insufficient to associate customers. Do not count untrusted account headers or freely replaceable cookies as verified identities.

Cloudflare implementation: the following Python 3 program builds a disabled rule for Enterprise Advanced Rate Limiting plus Bot Management. Its deployment-owned JSON input supplies host, path, nonempty observed asns and ja4_fingerprints, bot_score_below, period_seconds, requests_per_period and mitigation_seconds. Choose the sets from investigated traffic and calibrate thresholds on representative normal and abusive traffic; the script supplies no universal security thresholds. It matches low, evaluated bot scores together with an ASN set and a JA4 set. All matching combinations share a counter for that host/path within the provider's data-center scope, so rotating among the selected IPs, ASNs, UAs or fingerprints does not create a new counter there. This is one deployable restriction, not a cross-account attribution algorithm; previously unseen or changed features can escape its matching set.

# File: gateway/build_adaptive_rule.py
"""Build a disabled Cloudflare rule from an approved deployment policy; no API calls."""
from __future__ import annotations

import argparse
import json
import re
from pathlib import Path


def build_rule(policy: dict) -> dict:
    fields = {
        "host", "path", "asns", "ja4_fingerprints", "bot_score_below",
        "period_seconds", "requests_per_period", "mitigation_seconds",
    }
    if not isinstance(policy, dict) or set(policy) != fields:
        raise ValueError("supply exactly the documented deployment policy fields")
    host, path = policy["host"], policy["path"]
    if not isinstance(host, str) or not re.fullmatch(
        r"[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", host
    ):
        raise ValueError("host must be an exact lowercase ASCII hostname")
    if not isinstance(path, str) or not re.fullmatch(r"/[A-Za-z0-9/_~.%-]*", path):
        raise ValueError("path must be an exact API path without query or expression syntax")
    numeric = ("bot_score_below", "period_seconds", "requests_per_period", "mitigation_seconds")
    if any(type(policy[key]) is not int for key in numeric):
        raise ValueError("rate and score settings require integers, not booleans")
    # These are provider API constraints, not recommended deployment thresholds.
    if not 2 <= policy["bot_score_below"] <= 100:
        raise ValueError("bot_score_below must select from evaluated scores 1 through 99")
    if policy["period_seconds"] not in {10, 60, 120, 300, 600, 3600}:
        raise ValueError("unsupported Cloudflare rate period")
    if policy["mitigation_seconds"] not in {0, 10, 60, 120, 300, 600, 3600, 86400}:
        raise ValueError("unsupported Cloudflare mitigation timeout")
    if policy["requests_per_period"] <= 0:
        raise ValueError("requests_per_period must be positive")
    asns, fingerprints = policy["asns"], policy["ja4_fingerprints"]
    if not isinstance(asns, list) or not asns or any(
        type(value) is not int or not 1 <= value <= 4294967295 for value in asns
    ):
        raise ValueError("asns must be a nonempty list of valid ASN integers")
    if not isinstance(fingerprints, list) or not fingerprints or any(
        not isinstance(value, str) or not re.fullmatch(r"[a-z0-9_]{10,80}", value)
        for value in fingerprints
    ):
        raise ValueError("supply observed JA4 values, not expressions or empty fingerprints")
    asn_set = "{" + " ".join(str(value) for value in sorted(set(asns))) + "}"
    ja4_set = "{" + " ".join(json.dumps(value) for value in sorted(set(fingerprints))) + "}"
    expression = " and ".join((
        f"http.host eq {json.dumps(host)}",
        'http.request.method eq "POST"',
        f"http.request.uri.path eq {json.dumps(path)}",
        "cf.bot_management.score ge 1",
        f"cf.bot_management.score lt {policy['bot_score_below']}",
        "not cf.bot_management.verified_bot",
        f"ip.src.asnum in {asn_set}",
        f"cf.bot_management.ja4 in {ja4_set}",
    ))
    return {
        "description": "AI endpoint: aggregate a reviewed suspicious traffic set",
        "enabled": False,
        "action": "block",
        "expression": f"({expression})",
        "ratelimit": {
            "characteristics": ["cf.colo.id", "http.host", "http.request.uri.path"],
            "period": policy["period_seconds"],
            "requests_per_period": policy["requests_per_period"],
            "mitigation_timeout": policy["mitigation_seconds"],
        },
    }


def unique_keys(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate policy key: {key}")
        result[key] = value
    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("policy", type=Path)
    args = parser.parse_args()
    policy = json.loads(args.policy.read_text(encoding="utf-8"), object_pairs_hook=unique_keys)
    print(json.dumps(build_rule(policy), indent=2))

Save the approved input as rate-abuse-policy.json and run python gateway/build_adaptive_rule.py rate-abuse-policy.json to produce the rule JSON. Fetch the zone's http_ratelimit entry point with GET /client/v4/zones/{zone_id}/rulesets/phases/http_ratelimit/entrypoint; append the object with POST /client/v4/zones/{zone_id}/rulesets/{ruleset_id}/rules. Create the phase entry point only if absent. Read back the returned rule ID, expression, characteristics, action and disabled state; preserve existing rules and their ordering. A disabled rule does not evaluate traffic: preview its matching traffic in analytics, then use the provider's supported logging mode or a controlled staging deployment to measure matches before enforcement. Enable through the normal configuration deployment only after platform validation and replay. The builder validates input syntax, not product entitlements or live rule acceptance.

The example excludes Cloudflare score 0 (not evaluated) and empty/unmatched JA4 from this adaptive rule; the base guardrail still applies. Verified bots are excluded from this extra restriction only, not from base or budget limits. Validate legitimate machine clients that are not on a vendor's verified-bot list; grant any exception from independently verified service identity, not a self-declared UA. The rule applies the default rate-limit block response, HTTP 429. For browser endpoints, a separate supported challenge policy may be appropriate; do not require machine APIs to solve browser challenges.

5. Operate with bounded failure and recovery behavior

Keep short-lived restriction state separate from the base entitlement. Use approved thresholds and hysteresis or a minimum hold period supported by the platform to avoid oscillation; do not continuously rewrite rules or rotate rule IDs as scores fluctuate. Cloudflare mitigation_timeout limits a triggered mitigation; it does not expire the rule itself. Track the reviewed traffic set's validity in the deployment job, schedule renewal or disabling with the native rule API, and read back the resulting enabled state. Missing observations cannot renew the set or prove clean recovery. Disabling or expiring the extra rule must leave the base quota intact.

If optional bot or correlation telemetry is missing, stale or unavailable, retain the base guardrail and use the preselected conservative fallback for that route. Record degraded adaptive coverage rather than a clean score. Existing restrictions follow their configured hold/expiry and recovery rules; absence of new findings is not evidence to clear them early. Failure of base quota storage or its required authority remains an error that denies dispatch, not an ordinary quota-exceeded 429. Product rate limits can have propagation delay and region/data-center-local counters. In particular, Cloudflare's native rate counters are not globally shared, so retain application-side base limits and AID-I-003.003 budget reservations; do not claim a precise global request or financial cap from the edge rule.

6. Verify the actual restriction and normal-client impact

Use the deployed rule export/readback, edge mitigation logs and gateway dispatch/usage records to verify the same host, route, method, rule version and observation window. Reconcile edge rejections against absence of corresponding inference dispatch; requests not covered by the adaptive detector must remain visible. Measure matching coverage, escaped requests, false restrictions on legitimate clients, added latency and the provider's propagation/counting scope against the deployment policy. A successful API configuration write is not evidence that traffic was limited.

References: Akamai Bot Manager; Cloudflare Rulesets API; rate parameters; counter scope and semantics.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-I-003.002: Identity, Session & API Rate-Limit Throttling | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service