aiohttp 会话的 NTLM 身份验证

如何解决aiohttp 会话的 NTLM 身份验证

我正在尝试将具有大量 API 调用（即获取密钥列表的数据）的应用程序迁移到使用 asyncio，因为它是 IO 密集型任务。此 API 需要 NTLM 身份验证，因为它使用 Active Directory 凭据，为此我使用了以下代码：

session.auth = requests_ntlm.HttpNtlmAuth(username,password,session)

显然，asyncio 使用 aiohttp 进行异步会话处理。如此同步，它工作正常，但试图将其移动到更理想的异步/等待流，aiohttp 只接受基本身份验证凭据，如果 NTLM 身份验证传递给 TypeError: BasicAuth() tuple is required instead，则会抛出错误 aiohttp.ClientSession。以下是供参考的代码示例：

import asyncio
from aiohttp import ClientSession
from requests_ntlm import HttpNtlmAuth

async def fetch(url,session):
    async with session.get(url) as response:
        print(f"url: {url} ({response.status})")
        return await response.read()

async def run():
    url = "http://server/page/{}"
    tasks = []

    conn = aiohttp.TCPConnector(limit=10)
    async with ClientSession(connector=conn) as session:
        session.auth = HttpNtlmAuth(username,session)    # <--- Passing NTLM auth
        for page in range(100):
            task = asyncio.ensure_future(fetch(url.format(page),session))
            tasks.append(task)

        responses = await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run())
loop.run_until_complete(future)

有没有办法将 NTLM 凭据传递给 aiohttp 会话并使其工作？

解决方法

嗯，有两种方法可以做到这一点。

---

1 方法不是很好，因为它使用带有 loop.run_in_executor() 的异步任务。

def make_request(url,username,password):
    session = requests.Session()
    session.verify = False
    session.auth = HttpNtlmAuth(username,password)
    response = session.get(url)
    if response.status_code != 401:
        print("SUCCESS! You can login with: %s : %s" % (username,password))
        quit()
    else:
        print(username,password)


async def create_and_proceed(url_obj,password_data,username_data):
    tasks = []
    amount = 0
    requests_amount = 0
    loop = asyncio.get_event_loop()
    for user in username_data:
        for password in password_data:
            if amount == 50:
                await asyncio.gather(*tasks)
                amount = 0
                tasks = []
            tasks.append(loop.run_in_executor(None,make_request,url_obj,user,password))
            amount += 1
            requests_amount += 1
            print(f"Amount: {str(requests_amount)}",flush=True,end="\r")

---

两种方式更好，但我真的不知道它是否可行。

如果你能看到 HttpNtlmAuth 的源文件，你可以看到 HttpNtlmAuth 类是从 requests.auth.AuthBase() 继承的

class HttpNtlmAuth(AuthBase):
    """
    HTTP NTLM Authentication Handler for Requests.

    Supports pass-the-hash.
    """

    def __init__(self,password,session=None,send_cbt=True):
        """Create an authentication handler for NTLM over HTTP.

        :param str username: Username in 'domain\\username' format
        :param str password: Password
        :param str session: Unused. Kept for backwards-compatibility.
        :param bool send_cbt: Will send the channel bindings over a HTTPS channel (Default: True)
        """
        if ntlm is None:
            raise Exception("NTLM libraries unavailable")

        # parse the username
        try:
            self.domain,self.username = username.split('\\',1)
        except ValueError:
            self.username = username
            self.domain = ''

        if self.domain:
            self.domain = self.domain.upper()
        self.password = password
        self.send_cbt = send_cbt

        # This exposes the encrypt/decrypt methods used to encrypt and decrypt messages
        # sent after ntlm authentication. These methods are utilised by libraries that
        # call requests_ntlm to encrypt and decrypt the messages sent after authentication
        self.session_security = None

让我们看看 AuthBase() 的真正含义：

class AuthBase(object):
    """Base class that all auth implementations derive from"""

    def __call__(self,r):
        raise NotImplementedError('Auth hooks must be callable.')

所以如果我是对的，AuthBase() 类唯一要做的就是检查 Auth 钩子是否可调用。所以基本上你需要自己实现它...