NTLM authentication for aiohttp session

Question

I'm trying to migrate an application with a lot of API calls (i.e. fetching data for a list of keys) to use asyncio as it's an IO-intensive task. This API requires NTLM authentication as it uses Active Directory credentials and I was using below code for this purpose:

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

Apparently, asyncio uses aiohttp for asynchronous session handling. So synchronous, it works fine but trying to move it to a more desirable async/await flow, aiohttp only accepts basic auth credentials and would throw an error TypeError: BasicAuth() tuple is required instead if NTLM auth is passed to aiohttp.ClientSession. Here's the code sample for reference:

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)

Is there any way to pass NTLM credentials to aiohttp session and make it work?

Answer 1

Well, there is 2 ways to to this.

---

1 way is not really good, 'cause it's using async tasks with 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 way is better, but I don't really know if it could work.

If you able to watch source files of HttpNtlmAuth, you can see that HttpNtlmAuth class is inherted from 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

Let's see what is AuthBase() is really about:

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

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

So if I am right, the only thing AuthBase() class do - checks if Auth hooks are callable. So basicly you need to implement it with yourself...