ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > fastapi框架系列-HTTP 身份验证系统方案-HTTPDigest 摘要认证（需要自己实现，框架没提供实现逻辑）

fastapi框架系列-HTTP 身份验证系统方案-HTTPDigest 摘要认证（需要自己实现，框架没提供实现逻辑）

原创 数据库开发技术 作者：程序员小钟同学 时间：2022-04-06 14:54:00 0 删除 编辑

HTTP 身份验证系统方案-HTTPDigest 摘要认证

在基于标准的 HTTP 身份验证系统方案中包括：HTTPBasic 基本认证和HTTPDigest摘要认证。HTTPDigest是对Basic进一步的做的安全升级的优化机制，主要解决的问题避免通过网络传输密码。取而代之的是通过摘要随机参数取代密码。

其实简单的来说的，流程就是：

  • 1: 客户端发起请求需要认证的接口（无认证第一次访问接口的时候）
  • 2: 服务端回应返回服务端需要返回401 Unauthorized，并对客户端发出质询，需要输入用户名和密码等信息（1~2其实和basic差不多的，只是实现的机制不一样）
  • 3：客户端根据质询需要的信息，输入用户名和密码，客户端（浏览器）会在头部进行相关的MD5密码加密处理，然后传输(此时不再是明文传输密码了)
  • 4：然后服务端解析头部信息，根据用户名参数提取用户对于的密码，然后拿到密码之后，通过签名算法生成Ha1的算法，然后通过客户端提交头部中的response和自己通过签名算法算出的response（需要Ha1算法参与）是否一致来检测密码是否正确，来确保请求是否合法。
  • 请求合法则正常护理，不合法则继续发起质询，输入正确的用户名和密码。

当然HTTPDigest摘要认证是一个比较复杂的过程，不同的方式，生成的前面的算法不一样。

以下说明参考自：

https://blog.csdn.net/sinat_16643223/article/details/120247211

首先大概的了解一下HTTPDigest摘要认证需要关注的点有：

  • 质询头部中一些关键的参数说明：

    • WWW-Authentication ：用来定义使用何种方式（Basic、Digest、Bearer等）去进行认证以获取受保护的资源

    • realm 表示Web服务器中受保护文档的安全域（比如公司财务信息域和公司员工信息域），用来指示需要哪个域的用户名和密码

    • qop：保护质量，包含auth（默认的）和auth-int（增加了报文完整性检测）两种策略，【这个比较关键，这样底部不同的策略，后期response签名规则不一样】

    • nonce：服务端向客户端发送质询时附带的一个随机数，这个数会经常发生变化。客户端计算密码摘要时将其附加上去，使得多次生成同一用户的密码摘要各不相同，用来防止重放攻击

    • response：这是由用户代理软件计算出的一个字符串，以证明用户知道口令

其中上述的参数是比较用户的，下面我的结合一下摘要认证的过程实现Fastapi中一个摘要认证算法。首先因为这个摘要认证过程其实相对复杂，涉及的问题也多，所以我这里简化一下。

首先查看Fastpai提供的摘要认证源码：

class HTTPDigest(HTTPBase):
    def __init__(
        self,
        *,
        scheme_name: Optional[str] = None,
        description: Optional[str] = None,
        auto_error: bool = True,
    ):
        self.model = HTTPBaseModel(scheme="digest", description=description)
        self.scheme_name = scheme_name or self.__class__.__name__
        self.auto_error = auto_error

    async def __call__(
        self, request: Request
    ) -> Optional[HTTPAuthorizationCredentials]:
        authorization: str = request.headers.get("Authorization")
        scheme, credentials = get_authorization_scheme_param(authorization)
        if not (authorization and scheme and credentials):
            if self.auto_error:
                raise HTTPException(
                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
                )
            else:
                return None
        if scheme.lower() != "digest":
            raise HTTPException(
                status_code=HTTP_403_FORBIDDEN,
                detail="Invalid authentication credentials",
            )
        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)

通过查看源码发现好像，它是没有实现这个摘要认证过程中发起质询的，而只是简单的对头部信息一个校验而已，通过测试发现，其实却是也是，估计是它没考虑过实现这个吧！反正我是没看到怎么去实现。没办法只能自己去实现：

所以我们的可以依葫芦画瓢的，参考HttpBasic的实现自定义实现摘要认证一个简单的实现为：

def default_verify_password(password, credentials,request:Request):
    # 参数校验是否非法，如果非法的话，那么需要重新的发起一下需要重新的输入用户名称和密码

    realm = request.state.auth_seeions['realm']
    qop = request.state.auth_seeions['qop']
    nonce =request.state.auth_seeions['nonce']
    unauthorized_headers = {
        "WWW-Authenticate": f'Digest realm="{realm}",qop="{qop}",nonce="{nonce}'}

    datas = {item[0]: item[1] for item in
             [iten.split('=') for iten in credentials.replace('"', '').split(', ')]}
    # 验证值信息，验证账号密码信息，验证用户信息，验证报文信息
    if datas:
        ha1 = generate_ha1(username=datas.get('username'),realm=realm,password=password)
        ha2 = generate_ha2(request)
        # 计算摘要的公式MD5(MD5(A1):<nonce>:<nc>:<conce>:<qop>:MD5(A2))
        response = md5(
            f"{ha1}:{datas.get('nonce')}:{datas.get('nc')}:{datas.get('cnonce')}:{datas.get('qop')}:{ha2}".encode(
                'utf-8')).hexdigest()

        if response != datas.get('response'):
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED,
                detail="Unauthorized",
                headers=unauthorized_headers,
            )
        else:
            return True
    else:
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail="Unauthorized",
            headers=unauthorized_headers,
        )



def generate_ha1(username,realm, password):
    a1 = username + ":" + realm + ":" + password
    a1 = a1.encode('utf-8')
    return md5(a1).hexdigest()

def generate_ha2(requet: Request):
    a2 = requet.method + ":" + requet.url.path
    a2 = a2.encode('utf-8')
    return md5(a2).hexdigest()



class HTTPDigest(HTTPBase):

    def __init__(
            self,
            *,
            scheme_name: Optional[str] = None,
            description: Optional[str] = None,
            auto_error: bool = True,
            #  表示Web服务器中受保护文档的安全域（比如公司财务信息域和公司员工信息域），用来指示需要哪个域的用户名和密码
            realm: Optional[str] = None,
            # 保护质量，包含auth（默认的）和auth-int（增加了报文完整性检测）两种策略，（可以为空，但是）不推荐为空值
            qop: Optional[str] = "auth, auth-int",
            # 由服务器指定的字符串，客户端在后续指向同一个受保护区间的请求中应该在 Authorization 头中原样返回，建议使用 base64 或者 16 进制数
            opaque: Optional[str] = None,
    ):
        self.model = HTTPBaseModel(scheme="digest", description=description)
        self.scheme_name = scheme_name or self.__class__.__name__
        self.auto_error = auto_error
        self.realm=realm
        self.qop = qop
        self.opaque = opaque
        self.nonce = None
        self.random = SystemRandom()
        try:
            self.random.random()
        except NotImplementedError:  # pragma: no cover
            self.random = Random()

    def _generate_random(self):
        return md5(str(self.random.random()).encode('utf-8')).hexdigest()

    @property
    def default_generate_nonce(self):
        return self._generate_random()


    async def __call__(
            self, request: Request
    ):
        authorization: str = request.headers.get("Authorization")
        scheme, credentials = get_authorization_scheme_param(authorization)

        # 第一次请求，在没有进行认证或认证失败时，服务端需要返回401 Unauthorized，并对客户端发出质询，需要输入用户名和密码等信息
        self.request = request

        if not (authorization and scheme and credentials):
            if self.auto_error:
                if self.realm and self.qop:
                    self.nonce = self.default_generate_nonce
                    self.unauthorized_headers = {
                        "WWW-Authenticate": f'Digest realm="{self.realm}",qop="{self.qop}",nonce="{self.nonce}'}
                else:
                    self.unauthorized_headers = {"WWW-Authenticate": f'Digest realm="{self.realm}",'}
                raise HTTPException(
                    status_code=HTTP_401_UNAUTHORIZED,
                    detail="Unauthorized",
                    headers=self.unauthorized_headers,
                )
            else:
                return None

        # 如果认证的是已经通过了那么判断收是摘要认证的模式
        if scheme.lower() != "digest":
            raise HTTPException(
                status_code=HTTP_403_FORBIDDEN,
                detail="Invalid authentication credentials",
            )
        self.request.state.auth_seeions={}
        self.request.state.auth_seeions['realm']=self.realm
        self.request.state.auth_seeions['qop'] = self.qop
        self.request.state.auth_seeions['nonce'] = self.nonce
        # self

        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)

其中上述代码主要值得关注的点是：

  • ha1 ha2的算法生成
    • 当qop为auth或没指定方式的是，那么ha2的算法是：md5(a2)
    • 当qop为auth-init的时候，算法为：md5(a2:md5(body))
    • a1 算法生成是  username:realm:password
    • a2 算法是: metod:urlpath(uri) 请求方法名：请求路径
    • ha1 算法是: md5（a1）
    • ha2 算法分两种：
  • response 算法生成
    计算摘要的公式算法是：MD5(MD5(A1):<nonce>:<nc>:<conce>:<qop>:MD5(A2))
    • 当qop值为:auth或auth_init的时候
  • 质询发起时候的WWW-Authenticate的生成
      "WWW-Authenticate": f'Digest realm="{self.realm}",qop="{self.qop}",nonce="{self.nonce}'}
  • qop使用的方式为：auth

通过上面的方式实现后，我们的定义一个关于摘要认证的对象：

security = HTTPDigest(realm="nihao.baiwu.net", qop="auth")

然后定义我们的路由地址：


@app.get("/digest")
async def digest(request:Request,auth: HTTPAuthorizationCredentials = Depends(security)):
    # print(credentials)
    if default_verify_password(password='123456',credentials=auth.credentials,request=request):
        return PlainTextResponse('登入成功')

需要注意的点有：

response = md5(
            f"{ha1}:{datas.get('nonce')}:{datas.get('nc')}:{datas.get('cnonce')}:{datas.get('qop')}:{ha2}".encode(
                'utf-8')).hexdigest()

这个算法生成是针对于qop使用的auth的方式。

由于操作文档不支持这么模式的验证：

此时我们的可以通 浏览器访问我们的地址：

然后我们的输入错误的用户名和面则会一直弹窗要求输入，如果正确则通过。

甚至我们可以通过requests库来验证:

import requests
from requests.auth import HTTPDigestAuth

url = 'http://127.0.0.1:8000/digest'
resp = requests.get(url,auth=HTTPDigestAuth('admin','123456'))
print(resp.text)

上面仅仅是基本的实现，要扩展的东西很多，用户验证，随机数验证之类的。仅供参考。

总结

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： fastapi框架系列-再谈request body重复消费问题简析，推翻以前一些案例说明
下一篇： fastapi框架系列-HTTP 身份验证系统方案-HttpBasic 基本认证
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111688

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593