Negotiate authentication

.travis.yml

@@ -5,7 +5,7 @@ python:
   - "3.8"
   - "3.9"
 install:
-  - pip install .[testing]
+  - pip install .[testing,windows_auth]
 script:
   - pytest --cov=httpx_auth --cov-fail-under=100 --cov-report=term-missing
   - pip install idna==2.10

CHANGELOG.md

@@ -5,6 +5,15 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- `httpx_auth.authentication` contains a new `Negotiate` class that supports Kerberos and NTLM authentication without 
+  support for channel binding tokens
+- Added extra `windows_auth` to enable support for Negotiate and NTLM authentication
+
+### Changed
+- Optionally requires [`pyspnego[kerberos]`](https://github.com/jborean93/pyspnego)==0.1.6
+- Requires [`pytest`](https://docs.pytest.org/en/latest)==6.2.\* for testing
+- Requires [`pytest-mock`](https://github.com/pytest-dev/pytest-mock)==3.6.\* for testing
 
 ## [0.11.0] - 2021-08-19
 ### Changed

README.md

@@ -667,6 +667,51 @@ with httpx.Client() as client:
 | `username`              | User name.                     | Mandatory |               |
 | `password`              | User password.                 | Mandatory |               |
 
+## Negotiate and NTLM
+
+Support for Negotiate, Kerberos and NTLM authentication is optional, install with the `windows_auth` extra to enable
+this feature.
+
+You can use Negotiate, Kerberos and NTLM authentication with `httpx_auth.Negotiate`.
+
+### Using cached credentials for Kerberos authentication
+
+Cached credentials are used by default for Kerberos authentication, this relies on a Ticket-Granting Ticket being
+present on your system from `kinit` or similar. This is supported by default on Windows, on Linux it relies on system
+packages being installed. See documentation for the [pyspnego](https://pypi.org/project/pyspnego/) package for more
+information.
+
+```python
+import httpx
+from httpx_auth import Negotiate
+
+with httpx.Client() as client:
+    client.get('https://www.example.com', auth=Negotiate())
+```
+
+### Using provided credentials for NTLM authentication
+
+Where other credentials are required, or if Kerberos is not supported, provide a username and password for
+authentication:
+
+```python
+import httpx
+from httpx_auth import Negotiate
+
+with httpx.Client() as client:
+    client.get('https://www.example.com', auth=Negotiate('domain\\username', 'password'))
+```
+
+### Parameters
+
+| Name                    | Description                                                | Mandatory | Default value |
+|:------------------------|:-----------------------------------------------------------|:----------|:--------------|
+| `username`              | User name.                                                 | Optional  |               |
+| `password`              | User password.                                             | Optional  |               |
+| `force_ntlm`            | Force the use of NTLM auth                                 | Optional  | False         |
+| `service`               | Name portion of the server SPN                             | Optional  | "Service"     |
+| `max_redirects`         | Maximum number of redirects to follow while authenticating | Optional  | 10            |
+
 ## Multiple authentication at once
 
 You can also use a combination of authentication using `+`or `&`  as in the following sample:

httpx_auth/__init__.py

@@ -15,6 +15,7 @@
     OAuth2ClientCredentials,
     OktaClientCredentials,
     OAuth2ResourceOwnerPasswordCredentials,
+    Negotiate,
 )
 from httpx_auth.oauth2_tokens import JsonTokenFileCache
 from httpx_auth.aws import AWS4Auth

httpx_auth/authentication.py

@@ -2,11 +2,19 @@
 import os
 import uuid
 from hashlib import sha256, sha512
+from typing import Optional, Generator, List
 from urllib.parse import parse_qs, urlsplit, urlunsplit, urlencode
-from typing import Optional, Generator
 
 import httpx
 
+try:
+    import spnego
+
+    WINDOWS_AUTH = True
+except ImportError:
+    spnego = None
+    WINDOWS_AUTH = False
+
 from httpx_auth import oauth2_authentication_responses_server, oauth2_tokens
 from httpx_auth.errors import InvalidGrantRequest, GrantNotProvided
 
@@ -1165,6 +1173,185 @@ def __init__(self, username: str, password: str):
         httpx.BasicAuth.__init__(self, username, password)
 
 
+class Negotiate(httpx.Auth, SupportMultiAuth):
+    """
+    NOTE: This does not support Channel Bindings which can (and ought to be) supported by servers. This is due to a
+    limitation in the HTTPCore library at present.
+    """
+
+    _username: str
+    _password: str
+    force_ntlm: bool
+    auth_header: str
+    auth_complete: bool
+    auth_type: str
+    _service: str
+    _context_proxy: "spnego._context.ContextProxy"
+    max_redirects: int = 10
+
+    def __init__(
+        self,
+        username: str = None,
+        password: str = None,
+        force_ntlm: bool = False,
+        service: str = None,
+        max_redirects: int = 10,
+    ) -> None:
+        """
+        :param username: Username and domain (if required). Optional for servers that support Kerberos, required for
+        those that require NTLM
+        :param password: Password if required by server for authentication.
+        :param force_ntlm: Force authentication to use NTLM if available.
+        :param service: Service portion of the target Service Principal Name (default HTTP)
+        :return: None
+        """
+        if not WINDOWS_AUTH:
+            raise ImportError(
+                "Windows authentication support not enabled, install with the windows_auth extra."
+            )
+        if password and not username:
+            raise ValueError(
+                "Negotiate authentication with credentials requires username and password, no username was provided."
+            )
+        if force_ntlm and not (username and password):
+            raise ValueError(
+                "NTLM authentication requires credentials, provide a username and password."
+            )
+        self._username = username
+        self._password = password
+        self.force_ntlm = force_ntlm
+        self.auth_header = ""
+        self.auth_complete = False
+        self.auth_type = ""
+        self._service = service
+        self.max_redirects = max_redirects
+
+    def auth_flow(
+        self, request: httpx.Request
+    ) -> Generator[httpx.Request, httpx.Response, None]:
+
+        responses = []
+        response = yield request
+        responses.append(response)
+
+        redirect_count = 0
+
+        # If anything comes back except an authenticate challenge then return it for the client to deal with, hopefully
+        # a successful response.
+        if responses[-1].status_code != 401:
+            return responses[-1]
+
+        # Otherwise authenticate. Determine the authentication name to use, prefer Negotiate if available.
+        self.auth_type = self._auth_type_from_header(
+            responses[-1].headers.get("WWW-Authenticate")
+        )
+        if self.auth_type is None:
+            return responses[-1]
+
+        # Run authentication flow.
+        yield from self._do_auth_flow(request, responses)
+
+        # If we were redirected we will need to rerun the auth flow on the new url, repeat until either we receive a
+        # status that is not 401 Unauthorized, or until the url we ended up at is the same as the one we requested.
+        while responses[-1].status_code == 401 and responses[-1].url != request.url:
+            redirect_count += 1
+            if redirect_count > self.max_redirects:
+                raise httpx.TooManyRedirects(
+                    message=f"Redirected too many times ({self.max_redirects}).",
+                    request=request,
+                )
+            request.url = responses[-1].url
+            yield from self._do_auth_flow(request, responses)
+
+        return responses[-1]
+
+    def _do_auth_flow(
+        self, request: httpx.Request, responses: List[httpx.Response]
+    ) -> Generator[httpx.Request, httpx.Response, None]:
+        # Phase 1:
+        # Configure context proxy, generate message header, attach to request and resend.
+        host = request.url.host
+        self.context_proxy = self._new_context_proxy()
+        self.context_proxy.spn = "{0}/{1}".format(
+            self._service.upper() if self._service else "HTTP", host
+        )
+        request.headers["Authorization"] = self._make_authorization_header(
+            self.context_proxy.step(None)
+        )
+        response = yield request
+        responses.append(response)
+
+        # Phase 2:
+        # Server responds with Challenge message, parse the authenticate header and deal with cookies. Some web apps use
+        # cookies to store progress in the auth process.
+        if "set-cookie" in responses[-1].headers:
+            request.headers["Cookie"] = responses[-1].headers["Cookie"]
+
+        auth_header_bytes = self._parse_authenticate_header(
+            responses[-1].headers["WWW-Authenticate"]
+        )
+
+        # Phase 3:
+        # Generate Authenticate message, attach to the request and resend it. If the user is authorized then this will
+        # succeed. If not then this will fail.
+        self.auth_header = self._make_authorization_header(
+            self.context_proxy.step(auth_header_bytes)
+        )
+        request.headers["Authorization"] = self.auth_header
+        response = yield request
+        responses.append(response)
+
+    def _new_context_proxy(self) -> "spnego._context.ContextProxy":
+        client = spnego.client(
+            self._username,
+            self._password,
+            service=self._service,
+            protocol="ntlm" if self.force_ntlm else "negotiate",
+        )
+        if self.force_ntlm:
+            client.options = spnego.NegotiateOptions.use_ntlm
+            client.protocol = "ntlm"
+        return client
+
+    def _parse_authenticate_header(self, header: str) -> bytes:
+        """
+        Extract NTLM/Negotiate value from Authenticate header and convert to bytes
+        :param header: str WWW-Authenticate header
+        :return: bytes Negotiate challenge
+        """
+
+        auth_strip = self.auth_type.lower() + " "
+        auth_header_value = next(
+            s
+            for s in (val.lstrip() for val in header.split(","))
+            if s.lower().startswith(auth_strip)
+        )
+        return base64.b64decode(auth_header_value[len(auth_strip) :])
+
+    def _make_authorization_header(self, response_bytes: bytes) -> str:
+        """
+        Convert the auth bytes to base64 encoded string and build Authorization header.
+        :param response_bytes: bytes auth response content
+        :return: str Authorization/Proxy-Authorization header
+        """
+
+        auth_response = base64.b64encode(response_bytes).decode("ascii")
+        return f"{self.auth_type} {auth_response}"
+
+    @staticmethod
+    def _auth_type_from_header(header: str) -> Optional[str]:
+        """
+        Given a WWW-Authenticate header, returns the authentication type to use.
+        :param header: str Authenticate header
+        :return: Optional[str] Authentication type or None if not supported
+        """
+        if "negotiate" in header.lower():
+            return "Negotiate"
+        elif "ntlm" in header.lower():
+            return "NTLM"
+        return None
+
+
 class _MultiAuth(httpx.Auth):
     """Authentication using multiple authentication methods."""
 

setup.py

@@ -46,6 +46,12 @@
             "pytest_httpx==0.13.*",
             # Used to check coverage
             "pytest-cov==2.*",
+            # Used to test NTLM support
+            "pytest==6.*",
+            "pytest-mock==3.6.*"
+        ],
+        'windows_auth': [
+            "pyspnego[kerberos]==0.1.6"
         ]
     },
     python_requires=">=3.6",