from __future__ import annotations

import base64

import binascii

import sys

from dataclasses import dataclass

from .._helpers import is_ipv6_address

from .._version import __title__, __version__

from .errors import ReplyError

DEFAULT_USER_AGENT = (

f"Python/{sys.version_info[0]}.{sys.version_info[1]} {__title__}/{__version__}"

)

CRLF = "\r\n"

@dataclass

class BasicAuth:

login: str

password: str

encoding: str = "latin1"

def __post_init__(self) -> None:

if self.login is None:

raise ValueError("None is not allowed as login value")

if self.password is None:

raise ValueError("None is not allowed as password value")

if ":" in self.login:

raise ValueError('A ":" is not allowed in login (RFC 1945#section-11.1)')

@classmethod

def decode(cls, auth_header: str, encoding: str = "latin1") -> BasicAuth:

"""Create a BasicAuth object from an Authorization HTTP header."""

try:

auth_type, encoded_credentials = auth_header.split(" ", 1)

except ValueError:

raise ValueError("Could not parse authorization header.") # noqa: B904

if auth_type.lower() != "basic":

raise ValueError(f"Unknown authorization method {auth_type}")

try:

decoded = base64.b64decode(

encoded_credentials.encode("ascii"),

validate=True,

).decode(encoding)

except binascii.Error:

raise ValueError("Invalid base64 encoding.") # noqa: B904

try:

# RFC 2617 HTTP Authentication

# https://www.ietf.org/rfc/rfc2617.txt

# the colon must be present, but the username and password may be

# otherwise blank.

username, password = decoded.split(":", 1)

except ValueError:

raise ValueError("Invalid credentials.") # noqa: B904

return cls(login=username, password=password, encoding=encoding)

def encode(self) -> str:

"""Encode credentials."""

creds = f"{self.login}:{self.password}".encode(self.encoding)

return f"Basic {base64.b64encode(creds).decode(self.encoding)}"

class _Buffer:

def __init__(self, encoding: str = "utf-8") -> None:

self._encoding = encoding

self._buffer = bytearray()

def append_line(self, line: str = "") -> None:

if line:

self._buffer.extend(line.encode(self._encoding))

self._buffer.extend(CRLF.encode("ascii"))

def dumps(self) -> bytes:

return bytes(self._buffer)

@dataclass

class ConnectRequest:

host: str

port: int

username: str | None

password: str | None

def dumps(self) -> bytes:

buff = _Buffer()

# RFC 3986 § 3.2.2 / RFC 7230 § 5.4: a literal IPv6 address in an

# authority component must be enclosed in square brackets so the

# `:` between literal and port is unambiguous. Without brackets,

# `2001:db8::1:443` could be parsed as either the address

# `2001:db8::1` with port `443` or the address `2001:db8::1:443`

# with no port — proxies disagree, and several misroute the

# unbracketed form (typically returning `200 Connection

# established` against a connection that was never actually made).

host = f"[{self.host}]" if is_ipv6_address(self.host) else self.host

buff.append_line(f"CONNECT {host}:{self.port} HTTP/1.1")

buff.append_line(f"Host: {host}:{self.port}")

buff.append_line(f"User-Agent: {DEFAULT_USER_AGENT}")

if self.username and self.password:

auth = BasicAuth(self.username, self.password)

buff.append_line(f"Proxy-Authorization: {auth.encode()}")

buff.append_line()

return buff.dumps()

@dataclass

class ConnectReply:

status_code: int

message: str

@classmethod

def loads(cls, data: bytes) -> ConnectReply:

if not data:

raise ReplyError("Invalid proxy response") # pragma: no cover

line = data.split(CRLF.encode("ascii"), 1)[0]

line = line.decode("utf-8", "surrogateescape")

try:

_version, code, *reason = line.split()

except ValueError as e: # pragma: no cover

raise ReplyError(f"Invalid status line: {line}") from e

try:

status_code = int(code)

except ValueError as e: # pragma: no cover

raise ReplyError(f"Invalid status code: {code}") from e

status_message = " ".join(reason)

if status_code != 200: # noqa: PLR2004

msg = f"{status_code} {status_message}"

raise ReplyError(msg, error_code=status_code)

return cls(status_code=status_code, message=status_message)

class Connection:

def send(self, request: ConnectRequest) -> bytes:

return request.dumps()

def receive(self, data: bytes) -> ConnectReply:

return ConnectReply.loads(data)