Add browser-based authentication method (#375)

leplatrem · web-flow · commit cf0251e83fe8 · 2024-11-14T16:28:53.000+01:00
* Add browser-based authentication method

* Do not inspect type, just set the attr

* Add full end-to-end login test

* Be even more explicit in the test about reificiation
diff --git a/README.rst b/README.rst
@@ -100,6 +100,17 @@ the client must specify the expected type: ``kinto_http.BearerTokenAuth("XYPJTNs
 
     In other words, ``kinto_http.Client(auth="Bearer+OIDC XYPJTNsFKV2")`` is equivalent to ``kinto_http.Client(auth=kinto_http.BearerTokenAuth("XYPJTNsFKV2", type="Bearer+OIDC"))``
 
+Using the browser to authenticate via OAuth
+-------------------------------------------
+
+.. code-block:: python
+
+    import kinto_http
+
+    client = kinto_http.Client(server_url='http://localhost:8888/v1', auth=kinto_http.BrowserOAuth())
+
+The client will open a browser page and will catch the Bearer token obtained after the OAuth dance.
+
 
 Custom headers
 --------------
diff --git a/src/kinto_http/__init__.py b/src/kinto_http/__init__.py
@@ -10,12 +10,14 @@
     KintoBatchException,
     KintoException,
 )
+from kinto_http.login import BrowserOAuth
 from kinto_http.session import Session, create_session
 
 
 logger = logging.getLogger("kinto_http")
 
 __all__ = (
+    "BrowserOAuth",
     "BearerTokenAuth",
     "Endpoints",
     "Session",
diff --git a/src/kinto_http/client.py b/src/kinto_http/client.py
@@ -48,6 +48,12 @@ def __init__(
     ):
         self.endpoints = Endpoints()
 
+        try:
+            # See `BrowserOAuth` in login.py (for example).
+            auth.server_url = server_url
+        except AttributeError:
+            pass
+
         session_kwargs = dict(
             server_url=server_url,
             auth=auth,
diff --git a/src/kinto_http/login.py b/src/kinto_http/login.py
@@ -0,0 +1,90 @@
+import base64
+import json
+import threading
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import unquote
+
+import requests
+
+
+class RequestHandler(BaseHTTPRequestHandler):
+    def __init__(self, *args, set_jwt_token_callback=None, **kwargs):
+        self.set_jwt_token_callback = set_jwt_token_callback
+        super().__init__(*args, **kwargs)
+
+    def do_GET(self):
+        # Ignore non-auth requests (eg. favicon.ico).
+        if "/auth" not in self.path:  # pragma: no cover
+            self.send_response(404)
+            self.end_headers()
+            return
+
+        # Return a basic page to the user inviting them to close the page.
+        self.send_response(200)
+        self.send_header("Content-Type", "text/html")
+        self.end_headers()
+        self.wfile.write(
+            b"<html><body><h1>Login successful</h1>You can close this page.</body></html>"
+        )
+
+        # Decode the JWT token
+        encoded_jwt_token = unquote(self.path.replace("/auth/", ""))
+        decoded_data = base64.urlsafe_b64decode(encoded_jwt_token + "====").decode("utf-8")
+        jwt_data = json.loads(decoded_data)
+        self.set_jwt_token_callback(jwt_data)
+        # We don't want to stop the server immediately or it won't be
+        # able to serve the request response.
+        threading.Thread(target=self.server.shutdown).start()
+
+
+class BrowserOAuth(requests.auth.AuthBase):
+    def __init__(self, provider=None):
+        """
+        @param method: Name of the OpenID provider to get OAuth details from.
+        """
+        self.provider = provider
+        self.header_type = None
+        self.token = None
+
+    def set_jwt_token(self, jwt_data):
+        self.header_type = jwt_data["token_type"]
+        self.token = jwt_data["access_token"]
+
+    def __call__(self, r):
+        if self.token is not None:
+            r.headers["Authorization"] = "{} {}".format(self.header_type, self.token)
+            return r
+
+        # Fetch OpenID capabilities from the server root URL.
+        resp = requests.get(self.server_url + "/")
+        server_info = resp.json()
+        openid_info = server_info["capabilities"]["openid"]
+        if self.provider is None:
+            provider_info = openid_info["providers"][0]
+        else:
+            provider_info = [p for p in openid_info["providers"] if p["name"] == self.provider][0]
+
+        # Spawn a local server on a random port, in order to receive the OAuth dance
+        # redirection and JWT token content.
+        http_server = HTTPServer(
+            ("", 0),
+            lambda *args, **kwargs: RequestHandler(
+                *args, set_jwt_token_callback=self.set_jwt_token, **kwargs
+            ),
+        )
+        port = http_server.server_address[1]
+        redirect = f"http://localhost:{port}/auth/"
+        navigate_url = (
+            self.server_url
+            + provider_info["auth_path"]
+            + f"?callback={redirect}&scope=openid email"
+        )
+        webbrowser.open(navigate_url)
+
+        # Serve until the first request is received.
+        http_server.serve_forever()
+
+        # At this point JWT details were obtained.
+        r.headers["Authorization"] = "{} {}".format(self.header_type, self.token)
+        return r
diff --git a/tests/test_login.py b/tests/test_login.py
@@ -0,0 +1,105 @@
+import base64
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from unittest import mock
+from urllib.parse import parse_qs, quote, urlparse
+
+import pytest
+import requests
+
+from kinto_http.login import BrowserOAuth
+
+
+class RequestHandler(BaseHTTPRequestHandler):
+    def __init__(self, body, *args, **kwargs):
+        self.body = body
+        super().__init__(*args, **kwargs)
+
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.end_headers()
+        self.wfile.write(json.dumps(self.body).encode("utf-8"))
+
+
+@pytest.fixture
+def http_server():
+    rs_server = HTTPServer(
+        ("", 0),
+        lambda *args, **kwargs: RequestHandler(
+            {
+                "capabilities": {
+                    "openid": {
+                        "providers": [
+                            {
+                                "name": "other",
+                                "auth_path": "/openid/ldap/login",
+                            },
+                            {
+                                "name": "ldap",
+                                "auth_path": "/openid/ldap/login",
+                            },
+                        ]
+                    }
+                }
+            },
+            *args,
+            **kwargs,
+        ),
+    )
+    rs_server.port = rs_server.server_address[1]
+    threading.Thread(target=rs_server.serve_forever).start()
+
+    yield rs_server
+
+    rs_server.shutdown()
+
+
+@pytest.fixture
+def mock_oauth_dance():
+    def simulate_navigate(url):
+        """
+        Behave as the user going through the OAuth dance in the browser.
+        """
+        parsed = urlparse(url)
+        qs = parse_qs(parsed.query)
+        callback_url = qs["callback"][0]
+
+        token = {
+            "token_type": "Bearer",
+            "access_token": "fake-token",
+        }
+        json_token = json.dumps(token).encode("utf-8")
+        json_base64 = base64.urlsafe_b64encode(json_token)
+        encoded_token = quote(json_base64)
+        # This will open the local server started in `login.py`.
+        threading.Thread(target=lambda: requests.get(callback_url + encoded_token)).start()
+
+    with mock.patch("kinto_http.login.webbrowser") as mocked:
+        mocked.open.side_effect = simulate_navigate
+        yield
+
+
+def test_uses_first_openid_provider(mock_oauth_dance, http_server):
+    auth = BrowserOAuth()
+    auth.server_url = f"http://localhost:{http_server.port}/v1"
+
+    req = requests.Request()
+    auth(req)
+    assert "Bearer fake-token" in req.headers["Authorization"]
+
+    # Can be called infinitely and does not rely on remote server.
+    http_server.shutdown()
+    req = requests.Request()
+    auth(req)
+    assert "Bearer fake-token" in req.headers["Authorization"]
+
+
+def test_uses_specified_openid_provider(mock_oauth_dance, http_server):
+    auth = BrowserOAuth(provider="ldap")
+    auth.server_url = f"http://localhost:{http_server.port}/v1"
+
+    req = requests.Request()
+    auth(req)
+    assert "Bearer fake-token" in req.headers["Authorization"]
