Merge pull request #3 from zimedev/fix-webreq

fix webreq

certipy/commands/req.py

@@ -2,7 +2,6 @@
 import re
 from typing import List
 
-import requests
 from impacket.dcerpc.v5 import rpcrt
 from impacket.dcerpc.v5.dtypes import DWORD, LPWSTR, NULL, PBYTE, ULONG
 from impacket.dcerpc.v5.ndr import NDRCALL, NDRSTRUCT
@@ -11,8 +10,8 @@
 from impacket.dcerpc.v5.dcom.oaut import string_to_bin
 from impacket.dcerpc.v5.dcomrt import DCOMCALL, DCOMANSWER
 from impacket.uuid import uuidtup_to_bin
-from requests_ntlm import HttpNtlmAuth
-from urllib3 import connection
+import httpx
+from httpx_ntlm import HttpNtlmAuth
 
 from certipy.lib.certificate import (
     cert_id_to_parts,
@@ -45,20 +44,6 @@
 
 from .ca import CA
 
-
-def _http_request(self, method, url, body=None, headers=None):
-    if headers is None:
-        headers = {}
-    else:
-        # Avoid modifying the headers passed into .request()
-        headers = headers.copy()
-    super(connection.HTTPConnection, self).request(
-        method, url, body=body, headers=headers
-    )
-
-
-connection.HTTPConnection.request = _http_request
-
 MSRPC_UUID_ICPR = uuidtup_to_bin(("91ae6020-9e3c-11cf-8d7c-00aa00c091be", "0.0"))
 
 # https://winprotocoldoc.blob.core.windows.net/productionwindowsarchives/MS-WCCE/[MS-WCCE].pdf
@@ -458,7 +443,7 @@ def __init__(self, *args, **kwargs):
         self.base_url = ""
 
     @property
-    def session(self) -> requests.Session:
+    def session(self) -> httpx.Client:
         if self._session is not None:
             return self._session
 
@@ -476,23 +461,23 @@ def session(self) -> requests.Session:
 
         principal = "%s\\%s" % (self.target.domain, self.target.username)
 
-        session = requests.Session()
+        # Create a session with httpx
+        session = httpx.Client(auth=HttpNtlmAuth(principal, password))
         session.timeout = self.target.timeout
-        session.auth = HttpNtlmAuth(principal, password)
         session.verify = False
 
         base_url = "%s://%s:%i" % (scheme, self.target.target_ip, port)
         logging.info("Checking for Web Enrollment on %s" % repr(base_url))
 
-        session.headers["User-Agent"] = None
+        #session.headers["User-Agent"] = None
 
         success = False
         try:
             res = session.get(
                 "%s/certsrv/" % base_url,
                 headers={"Host": self.target.remote_name},
                 timeout=self.target.timeout,
-                allow_redirects=False,
+                follow_redirects=False,
             )
         except Exception as e:
             logging.warning("Failed to connect to Web Enrollment interface: %s" % e)
@@ -520,7 +505,7 @@ def session(self) -> requests.Session:
                     "%s/certsrv/" % base_url,
                     headers={"Host": self.target.remote_name},
                     timeout=self.target.timeout,
-                    allow_redirects=False,
+                    follow_redirects=False,
                 )
             except Exception as e:
                 logging.warning("Failed to connect to Web Enrollment interface: %s" % e)
@@ -708,7 +693,6 @@ def request(
 
         return self.retrieve(request_id)
 
-
 class Request:
     def __init__(
         self,

setup.py

@@ -21,7 +21,8 @@
         "dsinternals",
         "pyopenssl>=23.0.0",
         "requests",
-        "requests_ntlm",
+        "httpx",
+        "httpx-ntlm",
         'winacl; platform_system=="Windows"',
         'wmi; platform_system=="Windows"',
         "pycryptodome",