feat: add Discord and Bitbucket providers (#207)

* add discord and bitbucket provider

fix

Update fastapi_sso/sso/discord.py

Apply fix

Co-authored-by: Tomas Votava <info@tomasvotava.eu>

chore: resolve conflicts

* chore: remove unrelated changes

* chore: implement missing tests

---------

Co-authored-by: Afi <aconique@gmail.com>

Author: tomasvotava

.pre-commit-config.yaml

@@ -3,7 +3,7 @@ repos:
     hooks:
       - id: ruff
         name: ruff
-        entry: poe ruff
+        entry: poe ruff --fix
         language: system
         types: [python]
         pass_filenames: false

examples/bitbucket.py

@@ -0,0 +1,38 @@
+"""BitBucket Login Example
+"""
+
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi_sso.sso.bitbucket import BitbucketSSO
+
+CLIENT_ID = os.environ["CLIENT_ID"]
+CLIENT_SECRET = os.environ["CLIENT_SECRET"]
+
+app = FastAPI()
+
+sso = BitbucketSSO(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    redirect_uri="http://localhost:5000/auth/callback",
+    allow_insecure_http=True,
+)
+
+
+@app.get("/auth/login")
+async def auth_init():
+    """Initialize auth and redirect"""
+    with sso:
+        return await sso.get_login_redirect()
+
+
+@app.get("/auth/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.bitbucket:app", host="127.0.0.1", port=5000)

examples/discord.py

@@ -0,0 +1,38 @@
+"""Discord Login Example
+"""
+
+import os
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi_sso.sso.discord import DiscordSSO
+
+CLIENT_ID = os.environ["CLIENT_ID"]
+CLIENT_SECRET = os.environ["CLIENT_SECRET"]
+
+app = FastAPI()
+
+sso = DiscordSSO(
+    client_id=CLIENT_ID,
+    client_secret=CLIENT_SECRET,
+    redirect_uri="http://localhost:5000/auth/callback",
+    allow_insecure_http=True,
+)
+
+
+@app.get("/auth/login")
+async def auth_init():
+    """Initialize auth and redirect"""
+    with sso:
+        return await sso.get_login_redirect()
+
+
+@app.get("/auth/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.discord:app", host="127.0.0.1", port=5000)

fastapi_sso/__init__.py

@@ -4,6 +4,8 @@
 """
 
 from .sso.base import OpenID, SSOBase, SSOLoginError
+from .sso.bitbucket import BitbucketSSO
+from .sso.discord import DiscordSSO
 from .sso.facebook import FacebookSSO
 from .sso.fitbit import FitbitSSO
 from .sso.generic import create_provider
@@ -37,4 +39,6 @@
     "NotionSSO",
     "SpotifySSO",
     "TwitterSSO",
+    "BitbucketSSO",
+    "DiscordSSO",
 ]

fastapi_sso/sso/bitbucket.py

@@ -0,0 +1,60 @@
+"""BitBucket SSO Oauth Helper class"""
+
+from typing import TYPE_CHECKING, ClassVar, List, Optional, Union
+
+import pydantic
+
+from fastapi_sso.sso.base import DiscoveryDocument, OpenID, SSOBase
+
+if TYPE_CHECKING:
+    import httpx  # pragma: no cover
+
+
+class BitbucketSSO(SSOBase):
+    """Class providing login using BitBucket OAuth"""
+
+    provider = "bitbucket"
+    scope: ClassVar = ["account", "email"]
+    version = "2.0"
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
+        allow_insecure_http: bool = False,
+        scope: Optional[List[str]] = None,
+    ):
+        super().__init__(
+            client_id=client_id,
+            client_secret=client_secret,
+            redirect_uri=redirect_uri,
+            allow_insecure_http=allow_insecure_http,
+            scope=scope,
+        )
+
+    async def get_useremail(self, session: Optional["httpx.AsyncClient"] = None) -> dict:
+        """Get user email"""
+        if session is None:
+            raise ValueError("Session is required to make HTTP requests")
+
+        response = await session.get(f"https://api.bitbucket.org/{self.version}/user/emails")
+        return response.json()
+
+    async def get_discovery_document(self) -> DiscoveryDocument:
+        return {
+            "authorization_endpoint": "https://bitbucket.org/site/oauth2/authorize",
+            "token_endpoint": "https://bitbucket.org/site/oauth2/access_token",
+            "userinfo_endpoint": f"https://api.bitbucket.org/{self.version}/user",
+        }
+
+    async def openid_from_response(self, response: dict, session: Optional["httpx.AsyncClient"] = None) -> OpenID:
+        email = await self.get_useremail(session=session)
+        return OpenID(
+            email=email["values"][0]["email"],
+            display_name=response.get("display_name"),
+            provider=self.provider,
+            id=str(response.get("uuid")).strip("{}"),
+            first_name=response.get("nickname"),
+            picture=response.get("links", {}).get("avatar", {}).get("href"),
+        )

fastapi_sso/sso/discord.py

@@ -0,0 +1,56 @@
+"""Discord SSO Oauth Helper class"""
+
+from typing import TYPE_CHECKING, ClassVar, List, Optional, Union
+
+import pydantic
+
+from fastapi_sso.sso.base import DiscoveryDocument, OpenID, SSOBase
+
+if TYPE_CHECKING:
+    import httpx  # pragma: no cover
+
+
+class DiscordSSO(SSOBase):
+    """Class providing login using Discord OAuth"""
+
+    provider = "discord"
+    scope: ClassVar = ["identify", "email", "openid"]
+
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        redirect_uri: Optional[Union[pydantic.AnyHttpUrl, str]] = None,
+        allow_insecure_http: bool = False,
+        scope: Optional[List[str]] = None,
+    ):
+        super().__init__(
+            client_id=client_id,
+            client_secret=client_secret,
+            redirect_uri=redirect_uri,
+            allow_insecure_http=allow_insecure_http,
+            scope=scope,
+        )
+
+    async def get_discovery_document(self) -> DiscoveryDocument:
+        return {
+            "authorization_endpoint": "https://discord.com/oauth2/authorize",
+            "token_endpoint": "https://discord.com/api/oauth2/token",
+            "userinfo_endpoint": "https://discord.com/api/users/@me",
+        }
+
+    async def openid_from_response(self, response: dict, session: Optional["httpx.AsyncClient"] = None) -> OpenID:
+        user_id = response.get("id")
+        avatar = response.get("avatar")
+        picture = None
+        if user_id and avatar:
+            picture = f"https://cdn.discordapp.com/avatars/{user_id}/{avatar}.png"
+
+        return OpenID(
+            email=response.get("email"),
+            display_name=response.get("global_name"),
+            provider=self.provider,
+            id=user_id,
+            first_name=response.get("username"),
+            picture=picture,
+        )

tests/test_openid_responses.py

@@ -3,6 +3,7 @@
 import pytest
 
 from fastapi_sso.sso.base import OpenID, SSOBase
+from fastapi_sso.sso.discord import DiscordSSO
 from fastapi_sso.sso.facebook import FacebookSSO
 from fastapi_sso.sso.fitbit import FitbitSSO
 from fastapi_sso.sso.github import GithubSSO
@@ -219,6 +220,24 @@
             picture="https://avatars.yandex.net/get-yapic/123456/islands-200",
         ),
     ),
+    (
+        DiscordSSO,
+        {
+            "id": "test",
+            "avatar": "avatar",
+            "email": "test@example.com",
+            "global_name": "Test User",
+            "username": "testuser",
+        },
+        OpenID(
+            email="test@example.com",
+            first_name="testuser",
+            id="test",
+            picture="https://cdn.discordapp.com/avatars/test/avatar.png",
+            provider="discord",
+            display_name="Test User",
+        ),
+    ),
 )
 
 

tests/test_providers.py

@@ -8,6 +8,8 @@
 from utils import AnythingDict, Request, Response, make_fake_async_client
 
 from fastapi_sso.sso.base import OpenID, SecurityWarning, SSOBase
+from fastapi_sso.sso.bitbucket import BitbucketSSO
+from fastapi_sso.sso.discord import DiscordSSO
 from fastapi_sso.sso.facebook import FacebookSSO
 from fastapi_sso.sso.fitbit import FitbitSSO
 from fastapi_sso.sso.generic import create_provider
@@ -20,10 +22,10 @@
 from fastapi_sso.sso.microsoft import MicrosoftSSO
 from fastapi_sso.sso.naver import NaverSSO
 from fastapi_sso.sso.notion import NotionSSO
+from fastapi_sso.sso.seznam import SeznamSSO
 from fastapi_sso.sso.spotify import SpotifySSO
 from fastapi_sso.sso.twitter import TwitterSSO
 from fastapi_sso.sso.yandex import YandexSSO
-from fastapi_sso.sso.seznam import SeznamSSO
 
 GenericProvider = create_provider(
     name="generic",
@@ -52,6 +54,8 @@
     TwitterSSO,
     YandexSSO,
     SeznamSSO,
+    BitbucketSSO,
+    DiscordSSO,
 )
 
 # Run all tests for each of the listed providers

tests/test_providers_individual.py

@@ -1,6 +1,8 @@
+from unittest.mock import MagicMock
+
 import pytest
 
-from fastapi_sso import NotionSSO, OpenID, SSOLoginError
+from fastapi_sso import BitbucketSSO, NotionSSO, OpenID, SSOLoginError
 
 
 async def test_notion_openid_response():
@@ -23,3 +25,33 @@ async def test_notion_openid_response():
         await sso.openid_from_response(invalid_response)
     openid = OpenID(id="test", email="test@example.com", display_name="Test User", picture="avatar", provider="notion")
     assert await sso.openid_from_response(valid_response) == openid
+
+
+async def test_bitbucket_openid_response():
+    sso = BitbucketSSO("client_id", "client_secret")
+    valid_response = {
+        "uuid": "00000000-0000-0000-0000-000000000000",
+        "nickname": "testuser",
+        "links": {"avatar": {"href": "https://example.com/myavatar.png"}},
+        "display_name": "Test User",
+    }
+
+    class FakeSesssion:
+        async def get(self, url: str) -> MagicMock:
+            response = MagicMock()
+            response.json.return_value = {"values": [{"email": "test@example.com"}]}
+            return response
+
+    openid = OpenID(
+        id=valid_response["uuid"],
+        display_name=valid_response["display_name"],
+        provider="bitbucket",
+        email="test@example.com",
+        first_name="testuser",
+        picture=valid_response["links"]["avatar"]["href"],
+    )
+
+    with pytest.raises(ValueError, match="Session is required to make HTTP requests"):
+        await sso.openid_from_response(valid_response)
+
+    assert openid == await sso.openid_from_response(valid_response, FakeSesssion())