Implement Request.authorization

aiohttp/web_reqrep.py

@@ -17,7 +17,7 @@
 from urllib.parse import urlsplit, parse_qsl, unquote
 
 from . import hdrs
-from .helpers import reify
+from .helpers import reify, BasicAuth
 from .multidict import (CIMultiDictProxy,
                         CIMultiDict,
                         MultiDictProxy,
@@ -239,6 +239,22 @@ def if_modified_since(self, _IF_MODIFIED_SINCE=hdrs.IF_MODIFIED_SINCE):
                                          tzinfo=datetime.timezone.utc)
         return None
 
+    @reify
+    def authorization(self, _AUTHORIZATION=hdrs.AUTHORIZATION):
+        """Parse the Authorization header and return the username and password,
+        or ``None`` if the header is absent or cannot be parsed.
+
+        :rtype:  :class:`aiohttp.helpers.BasicAuth`
+
+        """
+        header = self.headers.get(_AUTHORIZATION)
+        if header is not None:
+            try:
+                return BasicAuth.decode(header)
+            except ValueError:
+                return None
+        return None
+
     @reify
     def keep_alive(self):
         """Is keepalive enabled by client?"""

tests/test_web_request.py

@@ -5,6 +5,7 @@
 from aiohttp.multidict import MultiDict, CIMultiDict
 from aiohttp.protocol import HttpVersion
 from aiohttp.protocol import RawRequestMessage
+from aiohttp.helpers import BasicAuth
 
 
 @pytest.fixture
@@ -192,6 +193,23 @@ def test_request_cookie__set_item(make_request):
         req.cookies['my'] = 'value'
 
 
+def test_request_authorization(make_request):
+    headers = CIMultiDict(AUTHORIZATION='Basic bmtpbTpwd2Q=')
+    req = make_request('GET', '/', headers=headers)
+    assert req.authorization == BasicAuth(login='nkim', password='pwd')
+
+
+def test_request_authorization_absent(make_request):
+    req = make_request('GET', '/')
+    assert req.authorization is None
+
+
+def test_request_authorization_invalid(make_request):
+    headers = CIMultiDict(AUTHORIZATION='Foobar')
+    req = make_request('GET', '/', headers=headers)
+    assert req.authorization is None
+
+
 def test_match_info(make_request):
     req = make_request('GET', '/')
     assert req.match_info is None