"""

requests.auth

~~~~~~~~~~~~~

This module contains the authentication handlers for Requests.

"""

from __future__ import annotations

import hashlib

import os

import re

import threading

import time

import warnings

from base64 import b64encode

from typing import TYPE_CHECKING, Any, Final, cast, overload

from ._internal_utils import to_native_string

from .compat import basestring, str, urlparse

from .cookies import extract_cookies_to_jar

from .utils import parse_dict_header

if TYPE_CHECKING:

    from http.cookiejar import CookieJar

    from typing import Any

    from .models import PreparedRequest, Response

CONTENT_TYPE_FORM_URLENCODED: Final = "application/x-www-form-urlencoded"

CONTENT_TYPE_MULTI_PART: Final = "multipart/form-data"

def _basic_auth_str(username: bytes | str, password: bytes | str) -> str:

    """Returns a Basic Auth string."""

    # "I want us to put a big-ol' comment on top of it that

    # says that this behaviour is dumb but we need to preserve

    # it because people are relying on it."

    #    - Lukasa

    #

    # These are here solely to maintain backwards compatibility

    # for things like ints. This will be removed in 3.0.0.

    if not isinstance(username, basestring):  # type: ignore[reportUnnecessaryIsInstance]  # runtime guard for non-str/bytes

        warnings.warn(

            "Non-string usernames will no longer be supported in Requests "

            f"3.0.0. Please convert the object you've passed in ({username!r}) to "

            "a string or bytes object in the near future to avoid "

            "problems.",

            category=DeprecationWarning,

        )

        username = str(username)

    if not isinstance(password, basestring):  # type: ignore[reportUnnecessaryIsInstance]  # runtime guard for non-str/bytes

        warnings.warn(

            "Non-string passwords will no longer be supported in Requests "

            f"3.0.0. Please convert the object you've passed in ({type(password)!r}) to "

            "a string or bytes object in the near future to avoid "

            "problems.",

            category=DeprecationWarning,

        )

        password = str(password)

    # -- End Removal --

    if isinstance(username, str):

        username = username.encode("latin1")

    if isinstance(password, str):

        password = password.encode("latin1")

    authstr = "Basic " + to_native_string(

        b64encode(b":".join((username, password))).strip()

    )

    return authstr

class AuthBase:

    """Base class that all auth implementations derive from"""

    def __call__(self, r: PreparedRequest) -> PreparedRequest:

        raise NotImplementedError("Auth hooks must be callable.")

class HTTPBasicAuth(AuthBase):

    """Attaches HTTP Basic Authentication to the given Request object."""

    username: bytes | str

    password: bytes | str

    @overload

    def __init__(self, username: str, password: str) -> None: ...

    @overload

    def __init__(self, username: bytes, password: bytes) -> None: ...

    def __init__(self, username: bytes | str, password: bytes | str) -> None:

        self.username = username

        self.password = password

    def __eq__(self, other: object) -> bool:

        return all(

            [

                self.username == getattr(other, "username", None),

                self.password == getattr(other, "password", None),

            ]

        )

    def __ne__(self, other: Any) -> bool:

        return not self == other

    def __call__(self, r: PreparedRequest) -> PreparedRequest:

        r.headers["Authorization"] = _basic_auth_str(self.username, self.password)

        return r

class HTTPProxyAuth(HTTPBasicAuth):

    """Attaches HTTP Proxy Authentication to a given Request object."""

    def __call__(self, r: PreparedRequest) -> PreparedRequest:

        r.headers["Proxy-Authorization"] = _basic_auth_str(self.username, self.password)

        return r

class HTTPDigestAuth(AuthBase):

    """Attaches HTTP Digest Authentication to the given Request object."""

    username: bytes | str

    password: bytes | str

    _thread_local: threading.local

    last_nonce: str

    nonce_count: int

    chal: dict[str, str]

    pos: int | None

    num_401_calls: int | None

    @overload

    def __init__(self, username: str, password: str) -> None: ...

    @overload

    def __init__(self, username: bytes, password: bytes) -> None: ...

    def __init__(self, username: bytes | str, password: bytes | str) -> None:

        self.username = username

        self.password = password

        # Keep state in per-thread local storage

        self._thread_local = threading.local()

    def init_per_thread_state(self) -> None:

        # Ensure state is initialized just once per-thread

        if not hasattr(self._thread_local, "init"):

            self._thread_local.init = True

            self._thread_local.last_nonce = ""

            self._thread_local.nonce_count = 0

            self._thread_local.chal = {}

            self._thread_local.pos = None

            self._thread_local.num_401_calls = None

    def build_digest_header(self, method: str, url: str) -> str | None:

        """

        :rtype: str

        """

        realm = self._thread_local.chal["realm"]

        nonce = self._thread_local.chal["nonce"]

        qop = self._thread_local.chal.get("qop")

        algorithm = self._thread_local.chal.get("algorithm")

        opaque = self._thread_local.chal.get("opaque")

        hash_utf8 = None

            _algorithm = "MD5"

        else:

            _algorithm = algorithm.upper()

        # lambdas assume digest modules are imported at the top level

        if _algorithm == "MD5" or _algorithm == "MD5-SESS":

            def md5_utf8(x: str | bytes) -> str:

                if isinstance(x, str):

                    x = x.encode("utf-8")

                return hashlib.md5(x, usedforsecurity=False).hexdigest()

            hash_utf8 = md5_utf8

        elif _algorithm == "SHA":

            def sha_utf8(x: str | bytes) -> str:

                if isinstance(x, str):

                    x = x.encode("utf-8")

                return hashlib.sha1(x, usedforsecurity=False).hexdigest()

            hash_utf8 = sha_utf8

        elif _algorithm == "SHA-256":

            def sha256_utf8(x: str | bytes) -> str:

                if isinstance(x, str):

                    x = x.encode("utf-8")

                return hashlib.sha256(x, usedforsecurity=False).hexdigest()

            hash_utf8 = sha256_utf8

        elif _algorithm == "SHA-512":

            def sha512_utf8(x: str | bytes) -> str:

                if isinstance(x, str):

                    x = x.encode("utf-8")

                return hashlib.sha512(x, usedforsecurity=False).hexdigest()

            hash_utf8 = sha512_utf8

        if hash_utf8 is None:

            return None

        def KD(s: str, d: str) -> str:

            return hash_utf8(f"{s}:{d}")

        # XXX not implemented yet

        entdig = None

        p_parsed = urlparse(url)

        #: path is request-uri defined in RFC 2616 which should not be empty

        path = p_parsed.path or "/"

        if p_parsed.query:

            path += f"?{p_parsed.query}"

        A1 = f"{self.username}:{realm}:{self.password}"

        A2 = f"{method}:{path}"

        HA1 = hash_utf8(A1)

        HA2 = hash_utf8(A2)

        if nonce == self._thread_local.last_nonce:

            self._thread_local.nonce_count += 1

        else:

            self._thread_local.nonce_count = 1

        ncvalue = f"{self._thread_local.nonce_count:08x}"

        s = str(self._thread_local.nonce_count).encode("utf-8")

        s += nonce.encode("utf-8")

        s += time.ctime().encode("utf-8")

        s += os.urandom(8)

        cnonce = hashlib.sha1(s, usedforsecurity=False).hexdigest()[:16]

        if _algorithm == "MD5-SESS":

            HA1 = hash_utf8(f"{HA1}:{nonce}:{cnonce}")  # type: ignore[reportConstantRedefinition]  # RFC 2617 terminology

        if not qop:

            respdig = KD(HA1, f"{nonce}:{HA2}")

        elif qop == "auth" or "auth" in qop.split(","):

            noncebit = f"{nonce}:{ncvalue}:{cnonce}:auth:{HA2}"

            respdig = KD(HA1, noncebit)

        else:

            # XXX handle auth-int.

            return None

        self._thread_local.last_nonce = nonce

        # XXX should the partial digests be encoded too?

        base = (

            f'username="{self.username}", realm="{realm}", nonce="{nonce}", '

            f'uri="{path}", response="{respdig}"'

        )

        if opaque:

            base += f', opaque="{opaque}"'

        if algorithm:

            base += f', algorithm="{algorithm}"'

        if entdig:

            base += f', digest="{entdig}"'

        if qop:

            base += f', qop="auth", nc={ncvalue}, cnonce="{cnonce}"'

        return f"Digest {base}"

    def handle_redirect(self, r: Response, **kwargs: Any) -> None:

        """Reset num_401_calls counter on redirects."""

        if r.is_redirect:

            self._thread_local.num_401_calls = 1

    def handle_401(self, r: Response, **kwargs: Any) -> Response:

        """

        Takes the given response and tries digest-auth, if needed.

        :rtype: requests.Response

        """

        # If response is not 4xx, do not auth

        # See https://github.com/psf/requests/issues/3772

        if not 400 <= r.status_code < 500:

            self._thread_local.num_401_calls = 1

            return r

        if self._thread_local.pos is not None:

            # Rewind the file position indicator of the body to where

            # it was to resend the request.

            if (seek := getattr(r.request.body, "seek", None)) is not None:

                seek(self._thread_local.pos)

        s_auth = r.headers.get("www-authenticate", "")

        if "digest" in s_auth.lower() and self._thread_local.num_401_calls < 2:

            self._thread_local.num_401_calls += 1

            pat = re.compile(r"digest ", flags=re.IGNORECASE)

            self._thread_local.chal = parse_dict_header(pat.sub("", s_auth, count=1))

            # Consume content and release the original connection

            # to allow our new request to reuse the same one.

            r.content

            r.close()

            prep = r.request.copy()

            cookie_jar = cast("CookieJar", prep._cookies)  # type: ignore[reportPrivateUsage]

            extract_cookies_to_jar(cookie_jar, r.request, r.raw)

            prep.prepare_cookies(cookie_jar)

            _digest_auth = self.build_digest_header(

                cast(str, prep.method), cast(str, prep.url)

            )

            if _digest_auth:

                prep.headers["Authorization"] = _digest_auth

            _r = r.connection.send(prep, **kwargs)

            _r.history.append(r)

            _r.request = prep

            return _r

        self._thread_local.num_401_calls = 1

        return r

    def __call__(self, r: PreparedRequest) -> PreparedRequest:

        # Initialize per-thread state, if needed

        self.init_per_thread_state()

        # If we have a saved nonce, skip the 401

        if self._thread_local.last_nonce:

            _digest_auth = self.build_digest_header(

                cast(str, r.method), cast(str, r.url)

            )

            if _digest_auth:

                r.headers["Authorization"] = _digest_auth

        if (tell := getattr(r.body, "tell", None)) is not None:

            self._thread_local.pos = tell()

        else:

            # In the case of HTTPDigestAuth being reused and the body of

            # the previous request was a file-like object, pos has the

            # file position of the previous body. Ensure it's set to

            # None.

            self._thread_local.pos = None

        r.register_hook("response", self.handle_401)

        r.register_hook("response", self.handle_redirect)

        self._thread_local.num_401_calls = 1

        return r

    def __eq__(self, other: object) -> bool:

        return all(

            [

                self.username == getattr(other, "username", None),

                self.password == getattr(other, "password", None),

            ]

        )

    def __ne__(self, other: Any) -> bool:

        return not self == other