Add support for Notion (#115)

* fix: apply additional headers to user endpoint
* feat: add support for Notion

.pylintrc

@@ -338,7 +338,7 @@ indent-after-paren=4
 indent-string='    '
 
 # Maximum number of characters on a single line.
-max-line-length=119
+max-line-length=120
 
 # Maximum number of lines in a module.
 max-module-lines=1000

README.md

@@ -50,6 +50,7 @@ I tend to process Pull Requests faster when properly caffeinated 😉.
 - Fitbit
 - Github (credits to [Brandl](https://github.com/Brandl) for hint using `accept` header)
 - generic (see [docs](https://tomasvotava.github.io/fastapi-sso/reference/sso.generic/))
+- Notion
 
 ### Contributed
 

examples/notion.py

@@ -0,0 +1,38 @@
+"""Github Login Example
+"""
+
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi_sso.sso.notion import NotionSSO
+
+CLIENT_ID = os.environ["CLIENT_ID"]
+CLIENT_SECRET = os.environ["CLIENT_SECRET"]
+
+app = FastAPI()
+
+sso = NotionSSO(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    redirect_uri="http://localhost:3000/oauth2/callback",
+    allow_insecure_http=True,
+)
+
+
+@app.get("/oauth2/login")
+async def auth_init():
+    """Initialize auth and redirect"""
+    with sso:
+        return await sso.get_login_redirect()
+
+
+@app.get("/oauth2/callback")
+async def auth_callback(request: Request):
+    """Verify login"""
+    with sso:
+        user = await sso.verify_and_process(request)
+        return user
+
+
+if __name__ == "__main__":
+    uvicorn.run(app="examples.notion:app", host="127.0.0.1", port=3000)

fastapi_sso/__init__.py

@@ -12,4 +12,5 @@
 from .sso.kakao import KakaoSSO
 from .sso.microsoft import MicrosoftSSO
 from .sso.naver import NaverSSO
+from .sso.notion import NotionSSO
 from .sso.spotify import SpotifySSO

fastapi_sso/sso/base.py

@@ -336,6 +336,7 @@ async def process_login(
         params = params or {}
         additional_headers = additional_headers or {}
         additional_headers.update(self.additional_headers or {})
+
         url = request.url
 
         if not self.allow_insecure_http and url.scheme != "https":
@@ -366,6 +367,7 @@ async def process_login(
             self.oauth_client.parse_request_body_response(json.dumps(content))
 
             uri, headers, _ = self.oauth_client.add_token(await self.userinfo_endpoint)
+            headers.update(additional_headers)
             session.headers.update(headers)
             response = await session.get(uri)
             content = response.json()

fastapi_sso/sso/notion.py

@@ -0,0 +1,35 @@
+"""Notion SSO Oauth Helper class"""
+
+from typing import TYPE_CHECKING, Optional
+
+from fastapi_sso.sso.base import DiscoveryDocument, OpenID, SSOBase, SSOLoginError
+
+if TYPE_CHECKING:
+    import httpx  # pragma: no cover
+
+
+class NotionSSO(SSOBase):
+    """Class providing login using Notion OAuth"""
+
+    provider = "notion"
+    scope = ["openid"]
+    additional_headers = {"Notion-Version": "2022-06-28"}
+
+    async def get_discovery_document(self) -> DiscoveryDocument:
+        return {
+            "authorization_endpoint": "https://api.notion.com/v1/oauth/authorize?owner=user",
+            "token_endpoint": "https://api.notion.com/v1/oauth/token",
+            "userinfo_endpoint": "https://api.notion.com/v1/users/me",
+        }
+
+    async def openid_from_response(self, response: dict, session: Optional["httpx.AsyncClient"] = None) -> OpenID:
+        owner = response["bot"]["owner"]
+        if owner["type"] != "user":
+            raise SSOLoginError(401, f"Notion login failed, owner is not a user but {response['bot']['owner']['type']}")
+        return OpenID(
+            id=owner["user"]["id"],
+            email=owner["user"]["person"]["email"],
+            picture=owner["user"]["avatar_url"],
+            display_name=owner["user"]["name"],
+            provider=self.provider,
+        )

tests/test_providers.py

@@ -19,6 +19,7 @@
 from fastapi_sso.sso.microsoft import MicrosoftSSO
 from fastapi_sso.sso.naver import NaverSSO
 from fastapi_sso.sso.spotify import SpotifySSO
+from fastapi_sso.sso.notion import NotionSSO
 
 GenericProvider = create_provider(
     name="generic",
@@ -41,6 +42,7 @@
     NaverSSO,
     SpotifySSO,
     GenericProvider,
+    NotionSSO,
 )
 
 # Run all tests for each of the listed providers

tests/test_providers_individual.py

@@ -0,0 +1,24 @@
+import pytest
+from fastapi_sso import NotionSSO, OpenID, SSOLoginError
+
+
+async def test_notion_openid_response():
+    sso = NotionSSO("client_id", "client_secret")
+    valid_response = {
+        "bot": {
+            "owner": {
+                "type": "user",
+                "user": {
+                    "id": "test",
+                    "person": {"email": "[email protected]"},
+                    "avatar_url": "avatar",
+                    "name": "Test User",
+                },
+            }
+        }
+    }
+    invalid_response = {"bot": {"owner": {"type": "workspace", "workspace": {}}}}
+    with pytest.raises(SSOLoginError):
+        await sso.openid_from_response(invalid_response)
+    openid = OpenID(id="test", email="[email protected]", display_name="Test User", picture="avatar", provider="notion")
+    assert await sso.openid_from_response(valid_response) == openid