Mitmproxy script to authenticate NTLM

mitmproxy-http-ntlm.py

from mitmproxy import http, ctx
from impacket.ntlm import getNTLMSSPType1, getNTLMSSPType3
import requests
import logging
import base64

username = "username"
password = "password"
domain = ''

class NTLMProxy:
    def __init__(self):
        pass

    def response(self, flow: http.HTTPFlow) -> None:
        if flow.response.headers.get('WWW-Authenticate'):
            logging.info('[+] Authentication requested')
            value = flow.response.headers['WWW-Authenticate']
            if 'NTLM' in value:
                logging.info('[+] NTLM requested')
                flow.response = self.send_auth_request(flow)
    
    def send_auth_request(self, flow: http.HTTPFlow) -> None:
        url = flow.request.url
        method = flow.request.method
        headers = flow.request.headers
        data = flow.request.text

        # This must be performed within the same session, otherwise authentication will fail!
        # I could not find a method to do this with mitmproxy its own flows, since replay.client will
        # not reuse the same connection
        session = requests.Session()

        # Create a negotiation NTLM Type 1
        ntlm_negotiate = getNTLMSSPType1()
        ntlm_negotiate_b64 = base64.b64encode(ntlm_negotiate.getData()).decode('ascii')
        headers.update({'Authorization': f'NTLM {ntlm_negotiate_b64}'})
        logging.info('[+] NTLM Negotiate')
        response = session.request(method=method, url=url, headers=headers, data=data, verify=False)

        # Extract the challenge from the server response; NTLM Type 2
        ntlm_challenge_b64 = response.headers.get('WWW-Authenticate')[5:]
        ntlm_challenge = base64.b64decode(ntlm_challenge_b64)

        # Generate the NTLM Type 3 message
        ntlm_authenticate, session_key = getNTLMSSPType3(
            type1=ntlm_negotiate, 
            type2=ntlm_challenge, 
            user=username, 
            password=password,
            domain=domain,
            use_ntlmv2=True)

        # Convert to Base64
        ntlm_authenticate_b64 = base64.b64encode(ntlm_authenticate.getData()).decode('ascii')

        # Make the authenticated request
        headers.update({'Authorization': f'NTLM {ntlm_authenticate_b64}'})
        logging.info('[+] NTLM Authenticate')
        response = session.request(method=method, url=url, headers=headers, data=data, verify=False)

        flow.response = http.Response.make(
            response.status_code,
            response.content,
            dict(response.headers)
        )

        return flow.response

addons = [
    NTLMProxy()
]