优化 Flask-Limiter：未认证用户请求的限流策略与处理

from flask import Flask, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import wraps

app = Flask(__name__)
limiter = Limiter(
    app=app,
    key_func=get_remote_address, # 根据远程IP地址进行限流
    default_limits=["1 per day", "1 per hour"], # 默认限流规则
    storage_uri="memory://", # 使用内存存储限流数据
)

# 模拟认证函数
def is_authenticated():
    # 在实际应用中，这里会根据 session、token 等进行认证判断
    return False # 假设用户未认证

@app.before_request
def check_rate_limit_globally():
    # 这里的逻辑可能导致问题：
    # 如果用户未认证，它可能不会显式返回，导致限流器仍然计数或生效
    print('--- 全局限流检查 ---')
    if is_authenticated():
        print('用户已认证，检查限流')
        resp = limiter.check() # 检查限流
        if resp and resp[1]:
            return jsonify({"message": "Rate limit exceeded"}), 429
    else:
        print('用户未认证')
        # 如果这里没有显式返回，请求会继续，限流器可能仍然工作

# 自定义认证装饰器
def authenticated_request(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not is_authenticated():
            print('路由装饰器检测到未认证')
            return jsonify({"message": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated_function

@app.route('/example')
@authenticated_request
def example_route():
    return jsonify({"message": "This is an example route"})

# if __name__ == '__main__':
#     app.run(debug=True)

from flask import Flask, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import wraps

app = Flask(__name__)
limiter = Limiter(
    app=app,
    key_func=get_remote_address, # 根据远程IP地址进行限流
    default_limits=["1 per day", "1 per hour"], # 默认限流规则
    storage_uri="memory://", # 使用内存存储限流数据
)

# 模拟认证函数
def is_authenticated():
    # 在实际应用中，这里会根据 session、token 等进行认证判断
    return False # 假设用户未认证

@app.before_request
def check_global_auth_and_rate_limit():
    """
    在所有请求处理前执行，优先检查认证状态。
    如果用户未认证，则直接返回 401，不再进行限流检查。
    如果用户已认证，则进行限流检查。
    """
    print('--- 检查全局认证和限流 ---')
    if not is_authenticated():
        # 用户未认证，立即返回 401 响应，阻止后续处理（包括限流计数）
        print('用户未认证，直接返回 401')
        return jsonify({"message": "Unauthorized"}), 401
    else:
        # 用户已认证，才进行限流检查
        print('用户已认证，检查限流')
        # 调用 limiter.check() 会触发限流逻辑并更新计数
        # 如果达到限流，则返回 429
        resp = limiter.check()
        if resp and resp[1]: # resp[1] 为 True 表示已超出限流
            print('已认证用户触发限流')
            return jsonify({"message": "Rate limit exceeded"}), 429
    print('--- 全局检查通过 ---')
    # 如果认证通过且未触发限流，则请求继续到路由处理器

# 自定义认证装饰器
def authenticated_request(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        # 理论上，如果 before_request 已经处理了未认证情况，
        # 这里的 is_authenticated() 应该总是返回 True。
        # 但作为安全冗余，保留此检查可以增加代码的健壮性。
        if not is_authenticated():
            print('路由装饰器检测到未认证 (冗余检查)')
            return jsonify({"message": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated_function

@app.route('/example')
@authenticated_request
def example_route():
    return jsonify({"message": "This is an example route"})

if __name__ == '__main__':
    app.run(debug=True)