Throw an exception in case of access denied (#138)

* Throw an exception in case of access denied * Update requests_ntlm/requests_ntlm.py Co-authored-by: Jordan Borean <[email protected]> * Minor tweak and add test for scenario --------- Co-authored-by: Jordan Borean <[email protected]>
requests · Oct 18, 2023 · 1488ba9 · 1488ba9
1 parent b80f1c7
commit 1488ba9
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 6 deletions.
diff --git a/requests_ntlm/requests_ntlm.py b/requests_ntlm/requests_ntlm.py
@@ -135,10 +135,16 @@ def retry_using_http_NTLM_auth(
         auth_strip = auth_type + " "
 
         ntlm_header_value = next(
-            s
-            for s in (val.lstrip() for val in auth_header_value.split(","))
-            if s.startswith(auth_strip)
-        ).strip()
+            (
+                s.strip()
+                for s in (val.lstrip() for val in auth_header_value.split(","))
+                if s.startswith(auth_strip)
+            ),
+            None,
+        )
+
+        if not ntlm_header_value:
+            raise PermissionError("Access denied: Server did not respond with NTLM challenge token")
 
         # Parse the challenge in the ntlm context and perform
         # the second step of authentication

diff --git a/tests/test_server.py b/tests/test_server.py
@@ -18,7 +18,11 @@ def negotiate_auth():
 def negotiate_and_ntlm_auth():
     return get_auth_response('NTLM', advertise_nego_and_ntlm=True)
 
-def get_auth_response(auth_type, advertise_nego_and_ntlm=False):
+@app.route("/no_challenge")
+def no_challenge():
+    return get_auth_response('Negotiate', no_challenge=True)
+
+def get_auth_response(auth_type, advertise_nego_and_ntlm=False, no_challenge=False):
     # Get the actual header that is returned by requests_ntlm
     actual_header = request.headers.get('Authorization', '')
 
@@ -43,7 +47,11 @@ def get_auth_response(auth_type, advertise_nego_and_ntlm=False):
         # Get the NTLM version number (bytes 9 - 12)
         message_type = struct.unpack("<I", msg[8:12])[0]
 
-        if message_type == negotiate_message_type:
+        if no_challenge:
+            response_headers = {'WWW-Authenticate': auth_type}
+            response = "access denied"
+            status_code = 401
+        elif message_type == negotiate_message_type:
             # Initial NTLM message from client, attach challenge token
             challenge_response = ('TlRMTVNTUAACAAAAAwAMADgAAAAzgoriASNFZ4mrze8AAAA'
                                   'AAAAAACQAJABEAAAABgBwFwAAAA9TAGUAcgB2AGUAcgACAA'

diff --git a/tests/unit/test_requests_ntlm.py b/tests/unit/test_requests_ntlm.py
@@ -54,6 +54,20 @@ def test_new_requests_are_used(self):
             self.assertTrue(res.history[0].request is not res.history[1].request)
             self.assertTrue(res.history[0].request is not res.request)
 
+    def test_permissions_denied(self):
+        auth = requests_ntlm.HttpNtlmAuth(
+            self.test_server_username,
+            self.test_server_password,
+        )
+
+        with self.assertRaises(PermissionError) as e:
+            requests.get(
+                url=f"{self.test_server_url}no_challenge",
+                auth=auth,
+            )
+
+        self.assertEqual(str(e.exception), "Access denied: Server did not respond with NTLM challenge token")
+
 
 class TestCertificateHash(unittest.TestCase):