Refactor http proto, bump version

python_socks/_protocols/http.py

@@ -1,7 +1,9 @@
 import sys
 from dataclasses import dataclass
+import base64
+import binascii
+from collections import namedtuple
 
-from .._basic_auth import BasicAuth
 from .._version import __title__, __version__
 
 from .errors import ReplyError
@@ -15,6 +17,58 @@
 CRLF = '\r\n'
 
 
+class BasicAuth(namedtuple('BasicAuth', ['login', 'password', 'encoding'])):
+    """Http basic authentication helper."""
+
+    def __new__(cls, login: str, password: str = '', encoding: str = 'latin1') -> 'BasicAuth':
+        if login is None:
+            raise ValueError('None is not allowed as login value')
+
+        if password is None:
+            raise ValueError('None is not allowed as password value')
+
+        if ':' in login:
+            raise ValueError('A ":" is not allowed in login (RFC 1945#section-11.1)')
+
+        # noinspection PyTypeChecker,PyArgumentList
+        return super().__new__(cls, login, password, encoding)
+
+    @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.')
+
+        if auth_type.lower() != 'basic':
+            raise ValueError('Unknown authorization method %s' % auth_type)
+
+        try:
+            decoded = base64.b64decode(encoded_credentials.encode('ascii'), validate=True).decode(
+                encoding
+            )
+        except binascii.Error:
+            raise ValueError('Invalid base64 encoding.')
+
+        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.')
+
+        # noinspection PyTypeChecker
+        return cls(username, password, encoding=encoding)
+
+    def encode(self) -> str:
+        """Encode credentials."""
+        creds = ('%s:%s' % (self.login, self.password)).encode(self.encoding)
+        return 'Basic %s' % base64.b64encode(creds).decode(self.encoding)
+
+
 class _Buffer:
     def __init__(self, encoding: str = 'utf-8'):
         self._encoding = encoding

python_socks/_version.py

@@ -1,2 +1,2 @@
 __title__ = 'python-socks'
-__version__ = '2.4.2'
+__version__ = '2.4.3'

tests/test_misc.py

@@ -2,7 +2,7 @@
 import pytest
 
 from python_socks._helpers import is_ip_address  # noqa
-from python_socks._basic_auth import BasicAuth  # noqa
+from python_socks._protocols.http import BasicAuth  # noqa
 
 
 @pytest.mark.parametrize('address', ('::1', b'::1', '127.0.0.1', b'127.0.0.1'))