Token Replay in Axum

Token replay is an authentication vulnerability where an attacker steals a valid session token (such as a JWT) and reuses it to impersonate the user. In Axum applications, this often occurs when tokens are stateless and lack mechanisms for invalidation. Axum's ergonomic extractors make it easy to implement authentication but can also lead to complacency: developers might rely solely on JWT signature and expiration validation, forgetting that a stolen token remains valid until it expires.

Axum's design as a minimalistic, composable framework encourages stateless architectures. Many Axum applications use JSON Web Tokens (JWTs) because they are easy to validate without server-side storage. A typical pattern is to extract the token from the Authorization header using an extractor or middleware, verify its signature and exp claim, and then trust the claims for authorization. However, this approach does not check whether the token has been used before or has been revoked.

Consider this vulnerable Axum middleware:

use axum::{middleware::Next, Request};
use jsonwebtoken::{decode, DecodingKey, Validation};

async fn auth_middleware(request: Request, next: Next) -> Result<Response, StatusCode> {
    let token = request.headers()
        .get("Authorization")
        .and_then(|h| h.to_str().ok())
        .and_then(|h| h.strip_prefix("Bearer "))
        .ok_or(StatusCode::UNAUTHORIZED)?;

    let claims: Claims = decode::<Claims>(
        token,
        &DecodingKey::from_secret(SECRET.as_bytes()),
        &Validation::default(),
    ).map_err(|_| StatusCode::UNAUTHORIZED)?.claims;

    // No replay protection: token is accepted indefinitely until expiration
    request.extensions_mut().insert(claims);
    Ok(next.run(request))
}

This middleware validates the token's signature and expiration but does not check any revocation list. If an attacker steals this token (e.g., via XSS, log leakage, or network sniffing), they can replay it to any endpoint protected by this middleware until the token expires. The attack is simple: capture the token from a legitimate request and include it in the Authorization: Bearer <token> header of a malicious request. Because the middleware only checks the signature and exp, the replayed token is accepted as valid.

The risk is amplified in Axum APIs that use long-lived JWTs (e.g., expiration set to days) or that do not implement logout functionality. Without a way to invalidate tokens, a stolen token is a permanent security breach until it naturally expires.

Detecting token replay vulnerabilities in Axum applications involves checking for the absence of token invalidation mechanisms. Manually, you would look for:

• Long JWT expiration times (e.g., days or weeks).
• No logout endpoint that invalidates the token.
• No server-side token blacklist or allowlist.
• Tokens that do not include a unique identifier (jti) claim.
• Tokens passed via URL query parameters, which are easily logged and replayed.

However, manual code review can miss subtle issues, especially in large codebases. This is where middleBrick helps. When you submit your Axum API URL to middleBrick, its Authentication check will:

1. Discover authentication endpoints (e.g., /login, /auth/token) from the OpenAPI spec or by probing common paths.
2. If any endpoint issues tokens without requiring prior authentication (a misconfiguration), middleBrick will attempt to obtain a token (e.g., by sending a default payload or exploiting a lack of credential validation).
3. It will then replay that token across multiple requests to other protected endpoints, including trying it in different contexts (e.g., as a query parameter if the API accepts tokens there).
4. If the token is accepted more than once, middleBrick flags a token replay vulnerability with a high severity score and provides a detailed report.

middleBrick's report includes a breakdown under the Authentication category, showing which endpoints are vulnerable, the specific requests that demonstrated replay, and step-by-step remediation guidance tailored to Axum. The scan takes only 5–15 seconds and requires no setup or credentials, making it easy to integrate into your development workflow.

Remediating token replay in Axum requires adding server-side token invalidation. The most robust approach is to maintain a token blacklist (or allowlist) that tracks revoked tokens. Since JWTs are stateless, you need a stateful component (e.g., Redis, database) to store blacklisted tokens until they expire. For Axum, this typically involves a custom extractor that checks the token's jti (JWT ID) against the blacklist before allowing the request.

Here's an implementation using a shared in-memory DashMap for demonstration (use a distributed cache like Redis in production):

use axum::{
    extract::{FromRequestParts, State},
    http::{request::Parts, StatusCode},
    response::{Response, IntoResponse},
};
use jsonwebtoken::{decode, DecodingKey, Validation, TokenData};
use std::sync::Arc;
use dashmap::DashMap;

#[derive(Debug, Clone)]
struct Claims {
    sub: String,
    exp: usize,
    jti: String, // Unique JWT ID
}

struct TokenBlacklist(Arc<DashMap<String, ()>>);

impl TokenBlacklist {
    fn new() -> Self {
        Self(Arc::new(DashMap::new()))
    }

    fn is_blacklisted(&self, jti: &str) -> bool {
        self.0.contains_key(jti)
    }

    fn blacklist(&self, jti: String) {
        self.0.insert(jti, ());
    }
}

#[derive(Debug)]
struct Auth(Claims);

#[async_trait::async_trait]
impl<S> FromRequestParts<S> for Auth
where
    S: Send + Sync,
{
    type Rejection = (StatusCode, &'static str);

    async fn from_request_parts(
        parts: &mut Parts,
        state: &S,
    ) -> Result<Self, Self::Rejection> {
        // Extract token from Authorization header
        let token = parts.headers
            .get("Authorization")
            .and_then(|h| h.to_str().ok())
            .and_then(|h| h.strip_prefix("Bearer "))
            .ok_or((StatusCode::UNAUTHORIZED, "Missing token"))?;

        // Get blacklist from request extensions (added via middleware)
        let blacklist = parts.extensions
            .get::<TokenBlacklist>()
            .ok_or((StatusCode::INTERNAL_SERVER_ERROR, "Blacklist not initialized"))?;

        // Validate token signature and expiration
        let token_data: TokenData<Claims> = decode::<Claims>(
            token,
            &DecodingKey::from_secret(SECRET.as_bytes()),
            &Validation::default(),
        ).map_err(|_| (StatusCode::UNAUTHORIZED, "Invalid token"))?;

        // Check if token is blacklisted
        if blacklist.is_blacklisted(&token_data.claims.jti) {
            return Err((StatusCode::UNAUTHORIZED, "Token revoked"));
        }

        Ok(Self(token_data.claims))
    }
}

// Logout endpoint that revokes the token
async fn logout(
    Auth(claims): Auth,
    extensions: axum::Extension<TokenBlacklist>,
) -> impl IntoResponse {
    extensions.0.blacklist(claims.jti);
    StatusCode::OK
}

To use this, add the blacklist to the request extensions via middleware so the Auth extractor can access it:

let blacklist = TokenBlacklist::new();
let app = Router::new()
    .route("/logout", post(logout))
    .layer(axum::middleware::from_fn(|request, next| {
        let blacklist = blacklist.clone();
        async move {
            request.extensions_mut().insert(blacklist);
            next.run(request).await
        }
    }));

Important: When issuing JWTs, include a unique jti claim (e.g., a UUID). This allows precise blacklisting. Also, set a reasonable expiration (e.g., 15 minutes) to limit the replay window even if a token is stolen. For APIs that cannot maintain a blacklist due to scalability, consider using very short-lived access tokens (5–15 minutes) with refresh tokens that can be rotated and revoked. However, this does not provide immediate invalidation.

After implementing these changes, rescan your API with middleBrick to verify that the token replay vulnerability is resolved. middleBrick will attempt to replay any obtained token and confirm it is rejected after revocation.