请求协助修改Python代码：基于State校验修复CSRF漏洞

我正在完成一项安全训练作业，需通过编码验证对CSRF漏洞修复的理解。训练所用代码沙盒仅提供红绿结果反馈，无其他调试信息。我并非寻求作业代做，而是希望获得修改Python代码的帮助，通过对比savedState与当前state值来修复CSRF漏洞。我对Python并不熟悉，虽理解CSRF修复原理，但无法完成正确的代码修改。以下是带有我尝试代码（已注释）的示例代码，当前版本无报错但未修复漏洞，恳请提供帮助、指导或可行代码。

import base64
import html
import json
import requests
import urllib


def encodeClientCredentials(clientId, clientSecret):
    s = urllib.parse.quote(clientId)+":" + urllib.parse.quote(clientSecret)
    return base64.b64encode(s.encode("utf-8"))


def clientCallback(query, clientData, authServerData, savedState):
    # Client
    # /GET /callback
    # query = {error, code, redirect_uri, state, scope, response_type }
    # savedState => Hash value saved by client after click "Approve"
    
    if "error" in query:
        return {
            "error": True,
            "msg": query["error"]
        }
    code = query["code"] if "code" in query else None
    # My code
    state = query["state"] if "state" in query else None
    # End of my Code
    form_data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": "http://client.com/callback"
    }
    encodedCredentials = encodeClientCredentials(
        clientData["client_id"],
        clientData["client_secret"]
    )
    encodedCredentials = encodedCredentials.decode("utf-8")
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": "Basic " + encodedCredentials
    }
    r = requests.post(
        authServerData["tokenEndpoint"],
        data=form_data,
        headers=headers
    )

    tokRes = r.text
    # My code
    #if state <> savedState:
    #    return {
    #        "error": True,
    #        "msg": "State value mismatch",
    #        "statusCode": r.status_code
    #    }
    # End of my code

    if r.status_code >= 200 and r.status_code < 300:
        body = json.loads(tokRes)
        scope = body["scope"]
        access_token = body["access_token"]
        return {
            "error": False,
            "access_token": access_token,
            "scope": scope,
            "statusCode": r.status_code
        }
    else:
        body = json.loads(tokRes)
        return {
            "error": True,
            "msg": body["error"],
            "statusCode": r.status_code
        }


def getContactXSS(access_token=None):
    if not access_token:
        return {
            "error": True,
            "msg": "No Access Token",
            "statusCode": 401
        }
    else:
        headers = {
            "Authorization": "Bearer " + access_token
        }
        r = requests.post(
            "http://protected:7012/contact-xss-get",
            headers=headers
        )
        rawResult = r.text
        if r.status_code >= 200 and r.status_code < 300:
            body = json.loads(rawResult)
            return {
                "error": False,
                "resource": body,
                "statusCode": r.status_code
            }
        else:
            return {
                "error": True,
                "statusCode": r.status_code,
            }

问题分析与修改方案 #

你的代码有两个关键问题导致CSRF漏洞未修复：

1. 校验时机错误：你把state校验放在了请求token之后，就算state不匹配，已经发起了token请求，无法阻止CSRF攻击。正确的做法是在发起任何敏感请求（比如获取token）之前就校验state。
2. 运算符错误：Python中判断不等于应该用!=，而不是<>（虽旧版本支持，但不推荐，且可能是代码失效的原因之一）。

另外，还要处理state不存在的情况——如果回调请求里没有state参数，直接判定为非法请求。

修改后的clientCallback函数如下：

def clientCallback(query, clientData, authServerData, savedState):
    # Client
    # /GET /callback
    # query = {error, code, redirect_uri, state, scope, response_type }
    # savedState => Hash value saved by client after click "Approve"
    
    if "error" in query:
        return {
            "error": True,
            "msg": query["error"]
        }
    
    # 先校验state，这是CSRF防护的核心步骤
    state = query.get("state")
    if not state or state != savedState:
        return {
            "error": True,
            "msg": "State value mismatch or missing",
            "statusCode": 403
        }
    
    code = query.get("code")
    if not code:
        return {
            "error": True,
            "msg": "Missing authorization code",
            "statusCode": 400
        }
    
    form_data = {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": "http://client.com/callback"
    }
    encodedCredentials = encodeClientCredentials(
        clientData["client_id"],
        clientData["client_secret"]
    )
    encodedCredentials = encodedCredentials.decode("utf-8")
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": "Basic " + encodedCredentials
    }
    r = requests.post(
        authServerData["tokenEndpoint"],
        data=form_data,
        headers=headers
    )

    tokRes = r.text

    if r.status_code >= 200 and r.status_code < 300:
        body = json.loads(tokRes)
        scope = body["scope"]
        access_token = body["access_token"]
        return {
            "error": False,
            "access_token": access_token,
            "scope": scope,
            "statusCode": r.status_code
        }
    else:
        body = json.loads(tokRes)
        return {
            "error": True,
            "msg": body["error"],
            "statusCode": r.status_code
        }

修改说明：

• 把state校验逻辑移到最前面，在处理code和发起token请求之前完成
• 使用query.get("state")更简洁地获取参数，同时判断state是否存在
• 用!=替代<>作为不等于运算符
• 给非法请求返回明确的错误信息和403状态码（表示权限不足/非法请求）
• 额外增加了code参数的校验，避免空code导致的无效请求

内容的提问来源于stack exchange，提问作者bregia