How do I detect and block attacks at runtime?

How do I detect and block attacks at runtime?

Detect the signal in the path of the action, then branch on it before the side effect. What is runtime application security? is the definition. The rest of this document is the pattern table and the instrumentation.

A scanner that reports after the query ran is monitoring. A proxy that copies the body to a second hop to classify it is a disclosure. Runtime detection is a function you call, in process, that returns enough to deny.

Runtime attack patterns

Four families show up in live traffic. They share a handler. They don't share a detector. The following table lists each family:

Injection on a form post and prompt injection on a chat box are the same placement (before the next hop) with different detectors. What is API abuse? covers the abuse family. Prevent data exfiltration through AI agents covers the data family. AI agent runtime security covers the agent family.

What to instrument

You need evidence at the handler, not after it.

Identity is the authenticated user, API key, or tenant. IP is a fallback, not the only key. Shared NATs and rotating clients make IP-only limits fail in both directions. Operation is the route, tool name, or job label. A search and a refund must not share one threshold. Payload shape is the parsed fields, not raw bytes only. Broken object-level authorization is an application question; for the HTTP baseline, see API security best practices. Free-text spans are the user messages, retrieved chunks, comments, and email bodies. Scan those. Don't scan opaque IDs.

Record the decision (allow or deny, rule, identity, correlation ID) so you can tune thresholds. A correlationId reconstructs a run. It doesn't change today's allow or deny.

Block without copying the body out

A common design is "send the request to a scanner, then proceed." That copies the body to a vendor so the vendor can tell you not to copy the body. Privacy review treats that vendor as a recipient.

Prefer a library in the process. Parse the request or the tool args in your handler. Call the detector on the strings you already have. Branch. On deny, return an HTTP 403 or 429 status code and don't call the database, the provider, or the outbound API.

Pattern matching, schema validation, and local sensitive-information detection can stay on-box. Shared counters (distributed rate limits) and some specialist models need a network call. That call must send a key and a verdict request, not a second copy of a medical note, unless you have accepted that disclosure.

For more information about the residency version of this argument, see keeping security inspection local.

Instrument in Node.js, Python, and Go

Extract identity and free text, run detectors, deny, then do the work. These handlers are generic. Swap in your own looksLikeInjection, tooMany, and hasSensitive functions.

Node.js:

1
export async function handle(req: Request, userId: string) {
2
  const body = await req.text();
3

4
  if (looksLikeInjection(body) || hasSensitive(body)) {
5
    // Generic body. Don't name the rule or echo the matched span.
6
    return new Response("Forbidden", { status: 403 });
7
  }
8

9
  const limit = await tooMany(userId);
10
  if (limit.exceeded) {
11
    return new Response("Too Many Requests", {
12
      status: 429,
13
      headers: { "retry-after": String(limit.resetSeconds) },
14
    });
15
  }
16

17
  return runBusinessLogic(body);
18
}

tooMany returns the reset time as well as the verdict, because a 429 without Retry-After turns a well-behaved client into a hot loop. Keep that counter in Redis or your rate-limit service, not in a module-level Map: a per-instance counter multiplies your intended limit by the number of instances, and it resets on every deploy.

Python (FastAPI):

1
from fastapi import Depends, FastAPI, HTTPException, Request
2

3
app = FastAPI()
4

5

6
@app.post("/v1/action")
7
async def action(
8
    request: Request,
9
    # From the session, not a query parameter the caller can rotate.
10
    user_id: str = Depends(current_user_id),
11
) -> dict[str, str]:
12
    body = (await request.body()).decode()
13
    if looks_like_injection(body) or has_sensitive(body):
14
        raise HTTPException(status_code=403, detail="Forbidden")
15

16
    limit = await too_many(user_id)
17
    if limit.exceeded:
18
        raise HTTPException(
19
            status_code=429,
20
            detail="Too Many Requests",
21
            headers={"retry-after": str(limit.reset_seconds)},
22
        )
23
    return await run_business_logic(body)

Go (net/http):

1
func action(w http.ResponseWriter, r *http.Request) {
2
  userID := userIDFromSession(r)
3

4
  // Cap the read. An unbounded io.ReadAll on a request body is its own
5
  // availability bug, and a detector on 200 MB of text is a timeout.
6
  body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
7
  if err != nil {
8
    http.Error(w, "bad request", http.StatusBadRequest)
9
    return
10
  }
11
  if looksLikeInjection(body) || hasSensitive(body) {
12
    http.Error(w, "forbidden", http.StatusForbidden)
13
    return
14
  }
15
  if limit := tooMany(r.Context(), userID); limit.Exceeded {
16
    w.Header().Set("Retry-After", strconv.Itoa(limit.ResetSeconds))
17
    http.Error(w, "too many requests", http.StatusTooManyRequests)
18
    return
19
  }
20
  runBusinessLogic(w, body)
21
}

Each handler reads identity from the session, bounds the body, screens the strings it already has, and returns before the work. Nothing here needs a new deployment topology.

On a path with no Request (a tool, an MCP handler, a queue job) the same predicates run on the arguments. There's nothing for a web application firewall (WAF) to inspect. For more information about that placement, see secure MCP server and agent tool calls.

One published SDK on the HTTP handler

The following Node.js protect() call is a web-attack filter, bot detection, and a sliding window on one route. guard() is a different API: no Request, and no bot primitive.

1
import arcjet, { detectBot, shield, slidingWindow } from "@arcjet/node";
2
import { createServer } from "node:http";
3

4
const aj = arcjet({
5
  key: process.env.ARCJET_KEY!,
6
  rules: [
7
    shield({ mode: "LIVE" }),
8
    detectBot({ mode: "LIVE", allow: [] }),
9
    slidingWindow({ mode: "LIVE", interval: 60, max: 100 }),
10
  ],
11
});
12

13
createServer(async (req, res) => {
14
  const decision = await aj.protect(req);
15
  if (decision.isDenied()) {
16
    res.writeHead(decision.reason.isRateLimit() ? 429 : 403);
17
    res.end();
18
    return;
19
  }
20
  res.writeHead(204);
21
  res.end();
22
}).listen(3000);

interval is seconds (or a duration string such as "1h"). max: 100 is a hundred hits a minute from the characteristics the adapter already extracts, which is ip.src unless you set characteristics on the client or the rule. Set it to ["userId"] and you have to pass that value on every protect() call, which is the point: a limit keyed on something the caller supplies isn't a limit.

Shield and sensitive-info detection run in process. Bot detection, prompt injection, and shared rate-limit counters need the Arcjet call. That split is what decides how much a network failure costs you on this route.

The same rules exist in the other SDKs with each language's own option names, so port the shape and check the reference rather than assuming the JavaScript field names. In Go the client is arcjet.NewClient, the rule is arcjet.SlidingWindow(arcjet.SlidingWindowOptions{Mode: arcjet.ModeLive, Interval: time.Minute, MaxRequests: 100}), and aj.Protect(ctx, r) returns a decision and an error. In Python the deny check is decision.is_denied() with the rule type on decision.reason_v2.type.

Dry-run first on a busy route. Move to live one rule at a time. For more information about in-code rollout, see enforce security rules at runtime in code. For more information about mapping those controls onto the OWASP GenAI list, see how to implement the OWASP Top 10 for LLM Applications.