[lichess] Start working on the Lichess integration

It will be a long run 😅, but hey that's the very first stone now laid.

With this 1st commit we add an ugly-but-functional Oauth2 flow that allows a user to get an API token from Lichess.

Author: olivierphi

.env.dist

@@ -1,3 +1,5 @@
 # This file will never be used in production, so it's ok to commit this secret key :-)
 SECRET_KEY=not-a-security-issue
 DATABASE_URL=sqlite:///db.sqlite3
+
+LICHESS_CLIENT_ID=zakuchess-local-dev

pyproject.toml

@@ -11,8 +11,8 @@ readme = "README.md"
 requires-python = ">=3.11"
 
 dependencies= [
+    # Django doesn't follow SemVer, so we need to specify the minor version:
     "Django==5.1.*",
-    # Django doesn't follow SemVer, so we need to specify the minor version
     "gunicorn==22.*",
     "django-alive==1.*",
     "chess==1.*",
@@ -22,9 +22,11 @@ dependencies= [
     "requests==2.*",
     "django-axes[ipware]==6.*",
     "whitenoise==6.*",
-    "django-import-export==3.*",
+    "django-import-export==4.*",
     "msgspec==0.18.*",
     "zakuchess",
+    "authlib==1.*",
+    "berserk>=0.13.2",
 ]
 
 

src/apps/lichess_bridge/__init__.py

@@ -0,0 +1,152 @@
+# Packages such as django-allauth or AuthLib do provide turnkey Django integrations
+# for OAuth2 - or even with Lichess specifically, for the former.
+# However, in my case I don't want to use Django's "auth" machinery to manage Lichess
+# users: all I want is to store in an HTTP-only cookie that they have attached
+# a Lichess account, alongside with the token we'll need to communicate with Lichess.
+# Hence, the following low level code, written after the Flask example given by Lichess:
+# https://github.com/lakinwecker/lichess-oauth-flask/blob/master/app.py
+# Authlib "vanilla Python" usage:
+# https://docs.authlib.org/en/latest/client/oauth2.html
+
+import functools
+from typing import TYPE_CHECKING, Literal
+
+import msgspec
+from authlib.common.security import generate_token
+from authlib.integrations.requests_client import OAuth2Session
+from django.conf import settings
+from django.urls import reverse
+
+if TYPE_CHECKING:
+    from typing import Self
+
+LICHESS_OAUTH2_SCOPES = ("board:play",)
+
+
+class LichessTokenRetrievalProcessContext(
+    msgspec.Struct,
+    kw_only=True,  # type: ignore[call-arg]
+):
+    """
+    Short-lived data required to complete the retrieval of an API token
+    from Lichess' OAuth2 process.
+    """
+
+    csrf_state: str
+    code_verifier: str
+    zakuchess_redirect_url: str  # an absolute HTTP/HTTPS URL
+
+    def to_cookie_content(self) -> str:
+        cookie_content = {
+            # We don't encode the redirect URL into the cookie, so let's customise
+            # what we need by encoding a dict, rather than "self"
+            "csrf": self.csrf_state,
+            "verif": self.code_verifier,
+        }
+        return msgspec.json.encode(cookie_content).decode()
+
+    @classmethod
+    def from_cookie_content(
+        cls,
+        cookie_content: str,
+        *,
+        zakuchess_hostname: str,
+        zakuchess_protocol: str = "https",
+    ) -> "Self":
+        cookie_content_dict = msgspec.json.decode(cookie_content)
+        redirect_uri = _get_lichess_oauth2_zakuchess_redirect_uri(
+            zakuchess_protocol,
+            zakuchess_hostname,
+        )
+
+        return cls(
+            csrf_state=cookie_content_dict["csrf"],
+            code_verifier=cookie_content_dict["verif"],
+            zakuchess_redirect_url=redirect_uri,
+        )
+
+    @classmethod
+    def create_afresh(
+        cls,
+        *,
+        zakuchess_hostname: str,
+        zakuchess_protocol: str = "https",
+    ) -> "Self":
+        """
+        Returns a context with randomly generated "CSRF state" and "code verifier".
+        """
+        redirect_uri = _get_lichess_oauth2_zakuchess_redirect_uri(
+            zakuchess_protocol, zakuchess_hostname
+        )
+
+        csrf_state = generate_token()
+        code_verifier = generate_token(48)
+
+        return cls(
+            csrf_state=csrf_state,
+            code_verifier=code_verifier,
+            zakuchess_redirect_url=redirect_uri,
+        )
+
+
+class LichessToken(msgspec.Struct):
+    token_type: Literal["Bearer"]
+    access_token: str
+    expires_in: int  # number of seconds
+    expires_at: int  # a Unix timestamp
+
+
+def get_lichess_token_retrieval_via_oauth2_process_starting_url(
+    *,
+    context: LichessTokenRetrievalProcessContext,
+) -> str:
+    lichess_authorization_endpoint = f"{settings.LICHESS_HOST}/oauth"
+
+    client = _get_lichess_client()
+    uri, state = client.create_authorization_url(
+        lichess_authorization_endpoint,
+        response_type="code",
+        state=context.csrf_state,
+        redirect_uri=context.zakuchess_redirect_url,
+        code_verifier=context.code_verifier,
+    )
+    assert state == context.csrf_state
+
+    return uri
+
+
+def extract_lichess_token_from_oauth2_callback_url(
+    *,
+    authorization_callback_response_url: str,
+    context: LichessTokenRetrievalProcessContext,
+) -> LichessToken:
+    lichess_token_endpoint = f"{settings.LICHESS_HOST}/api/token"
+
+    client = _get_lichess_client()
+    token_as_dict = client.fetch_token(
+        lichess_token_endpoint,
+        authorization_response=authorization_callback_response_url,
+        redirect_uri=context.zakuchess_redirect_url,
+        code_verifier=context.code_verifier,
+    )
+
+    return LichessToken(
+        **token_as_dict,
+    )
+
+
+@functools.lru_cache
+def _get_lichess_oauth2_zakuchess_redirect_uri(
+    zakuchess_protocol: str, zakuchess_hostname: str
+) -> str:
+    return f"{zakuchess_protocol}://{zakuchess_hostname}" + reverse(
+        "lichess_bridge:oauth2_token_callback"
+    )
+
+
+def _get_lichess_client() -> OAuth2Session:
+    return OAuth2Session(
+        client_id=settings.LICHESS_CLIENT_ID,
+        code_challenge_method="S256",
+        scope=" ".join(LICHESS_OAUTH2_SCOPES),
+    )

src/apps/lichess_bridge/authentication.py

@@ -0,0 +1,35 @@
+from typing import TYPE_CHECKING
+
+from dominate.tags import section
+from dominate.util import raw
+
+from apps.lichess_bridge.authentication import (
+    get_lichess_token_retrieval_via_oauth2_process_starting_url,
+)
+from apps.webui.components.layout import page
+
+if TYPE_CHECKING:
+    from django.http import HttpRequest
+
+    from apps.lichess_bridge.authentication import (
+        LichessTokenRetrievalProcessContext,
+    )
+
+
+def lichess_no_account_linked_page(
+    *,
+    request: "HttpRequest",
+    lichess_oauth2_process_context: "LichessTokenRetrievalProcessContext",
+) -> str:
+    target_url = get_lichess_token_retrieval_via_oauth2_process_starting_url(
+        context=lichess_oauth2_process_context
+    )
+
+    return page(
+        section(
+            raw(f"""Click here: <a href="{target_url}">{target_url}</a>"""),
+            cls="text-slate-50",
+        ),
+        request=request,
+        title="Lichess - no account linked",
+    )

src/apps/lichess_bridge/components/__init__.py

@@ -0,0 +1,61 @@
+import logging
+from typing import TYPE_CHECKING
+
+from msgspec import MsgspecError
+
+from .authentication import LichessTokenRetrievalProcessContext
+
+if TYPE_CHECKING:
+    from django.http import HttpRequest, HttpResponse
+
+_OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_NAME = "lichess.oauth2.ctx"
+# One day should be more than enough to let the user grant their authorisation:
+_OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_MAX_AGE = 3600 * 24
+
+
+_logger = logging.getLogger(__name__)
+
+
+def store_oauth2_token_retrieval_context_in_response_cookie(
+    *, context: LichessTokenRetrievalProcessContext, response: "HttpResponse"
+) -> None:
+    """
+    Store OAuth2 token retrieval context into a response cookie.
+    """
+
+    response.set_cookie(
+        _OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_NAME,
+        context.to_cookie_content(),
+        max_age=_OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_MAX_AGE,
+        httponly=True,
+    )
+
+
+def get_oauth2_token_retrieval_context_from_request(
+    request: "HttpRequest",
+) -> LichessTokenRetrievalProcessContext | None:
+    """
+    Returns a context created from the "CSRF state" and "code verifier" found in the request's cookie.
+    """
+    cookie_content: str | None = request.COOKIES.get(
+        _OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_NAME
+    )
+    if not cookie_content:
+        return None
+
+    try:
+        context = LichessTokenRetrievalProcessContext.from_cookie_content(
+            cookie_content,
+            zakuchess_hostname=request.get_host(),
+            zakuchess_protocol=request.scheme,
+        )
+        return context
+    except MsgspecError:
+        _logger.exception("Could not decode cookie content.")
+        return None
+
+
+def delete_oauth2_token_retrieval_context_from_cookies(
+    response: "HttpResponse",
+) -> None:
+    response.delete_cookie(_OAUTH2_TOKEN_RETRIEVAL_CONTEXT_COOKIE_NAME)

src/apps/lichess_bridge/components/pages/__init__.py

@@ -0,0 +1,14 @@
+from django.urls import path
+
+from . import views
+
+app_name = "lichess_bridge"
+
+urlpatterns = [
+    path("", views.lichess_home, name="homepage"),
+    path(
+        "webhook/oauth2/token-callback/",
+        views.lichess_webhook_oauth2_token_callback,
+        name="oauth2_token_callback",
+    ),
+]

src/apps/lichess_bridge/components/pages/lichess.py

@@ -0,0 +1,61 @@
+from typing import TYPE_CHECKING
+
+from django.http import HttpResponse
+from django.shortcuts import redirect
+from django.views.decorators.http import require_GET
+
+from .authentication import (
+    LichessTokenRetrievalProcessContext,
+    extract_lichess_token_from_oauth2_callback_url,
+)
+from .components.pages.lichess import lichess_no_account_linked_page
+from .cookie_helpers import (
+    delete_oauth2_token_retrieval_context_from_cookies,
+    get_oauth2_token_retrieval_context_from_request,
+    store_oauth2_token_retrieval_context_in_response_cookie,
+)
+
+if TYPE_CHECKING:
+    from django.http import HttpRequest
+
+
+def lichess_home(request: "HttpRequest") -> HttpResponse:
+    lichess_oauth2_process_context = LichessTokenRetrievalProcessContext.create_afresh(
+        zakuchess_hostname=request.get_host(),
+        zakuchess_protocol=request.scheme,
+    )
+
+    response = HttpResponse(
+        lichess_no_account_linked_page(
+            request=request,
+            lichess_oauth2_process_context=lichess_oauth2_process_context,
+        )
+    )
+    # We will need to re-use some of this context's data in the webhook below:
+    # --> let's store that in an HTTP-only cookie
+    store_oauth2_token_retrieval_context_in_response_cookie(
+        context=lichess_oauth2_process_context, response=response
+    )
+
+    return response
+
+
+@require_GET
+def lichess_webhook_oauth2_token_callback(request: "HttpRequest") -> HttpResponse:
+    # Retrieve a context from the HTTP-only cookie we created above:
+    lichess_oauth2_process_context = get_oauth2_token_retrieval_context_from_request(
+        request
+    )
+    if lichess_oauth2_process_context is None:
+        # TODO: Do something with that error
+        return redirect("lichess_bridge:homepage")
+
+    token = extract_lichess_token_from_oauth2_callback_url(
+        authorization_callback_response_url=request.get_full_path(),
+        context=lichess_oauth2_process_context,
+    )
+
+    response = HttpResponse(f"{token=}")
+    delete_oauth2_token_retrieval_context_from_cookies(response)
+
+    return response