aiohttp 会话的 NTLM 身份验证

Question

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

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

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

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, password, 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 会话并使其工作？

Answer 1

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

---

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")

---

2方式更好，但我真的不知道它是否可以工作。

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

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

    Supports pass-the-hash.
    """

    def __init__(self, username, 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 挂钩是否可调用。所以基本上你需要自己实现它......