ENG-3631: added azure auth

azure-auth/.gitignore

@@ -0,0 +1,5 @@
+*.db
+*.py[cod]
+.web
+__pycache__/
+assets/external/

azure-auth/azure_auth/.gitignore

@@ -0,0 +1,4 @@
+*.db
+*.py[cod]
+.web
+__pycache__/

azure-auth/azure_auth/auth/core.py

@@ -0,0 +1,76 @@
+import msal
+import reflex as rx
+from typing import Dict, List, Optional
+
+client_id: str = "0df2a88e-fddb-4cc2-b3e0-f475f162b373"
+client_secret: str = ""
+tenant_id: str = "f2c9cbbe-006b-46b8-9ad0-d877d8446d6d"
+authority = f"https://login.microsoftonline.com/{tenant_id}"
+login_redirect = "/"
+cache = msal.TokenCache()
+
+sso_app: msal.ClientApplication
+if client_secret:
+    sso_app = msal.ConfidentialClientApplication(
+        client_id=client_id,
+        client_credential=client_secret,
+        authority=authority,
+        token_cache=cache,
+    )
+else:
+    sso_app = msal.PublicClientApplication(
+        client_id=client_id,
+        authority=authority,
+        token_cache=cache,
+    )
+
+
+class State(rx.State):
+    _access_token: str = ""
+    _flow: dict
+    _token: Dict[str, str] = {}
+
+    def redirect_sso(self, scope: Optional[List] = None) -> rx.Component:
+        if scope is None:
+            scope = []
+        self._flow = sso_app.initiate_auth_code_flow(
+            scopes=scope, redirect_uri=f"{self.router.page.host}/callback"
+        )
+        return rx.redirect(self._flow["auth_uri"])
+
+    def require_auth(self):
+        if not self._token:
+            return self.redirect_sso()
+
+    @rx.var(cache=True)
+    def check_auth(self) -> bool:
+        return True if self._token else False
+
+    @rx.var(cache=True)
+    def token(self) -> Dict[str, str]:
+        return self._token
+
+    def logout(self):
+        self._token = {}
+        return rx.redirect(authority + "/oauth2/v2.0/logout")
+
+    def callback(self):
+        query_components = self.router.page.params
+
+        auth_response = {
+            "code": query_components.get("code"),
+            "client_info": query_components.get("client_info"),
+            "state": query_components.get("state"),
+            "session_state": query_components.get("session_state"),
+            "client-secret": client_secret,
+        }
+        try:
+            result = sso_app.acquire_token_by_auth_code_flow(
+                self._flow, auth_response, scopes=[]
+            )
+        except Exception as e:
+            return rx.toast(f"error something went wrong")
+        # this can be used for accessing graph
+        self._access_token = result.get("access_token")
+        self._token = result.get("id_token_claims")
+        return rx.redirect(login_redirect)

azure-auth/azure_auth/azure_auth.py

@@ -0,0 +1,15 @@
+"""Welcome to Reflex! This file outlines the steps to create a basic app."""
+
+import reflex as rx
+
+from rxconfig import config
+from azure_auth.pages import callback, home, logout
+
+
+class State(rx.State):
+    """The app state."""
+
+    ...
+
+
+app = rx.App()

azure-auth/azure_auth/pages/__init__.py

@@ -0,0 +1,28 @@
+import reflex as rx
+from azure_auth.auth.core import State as SsoState
+
+
+@rx.page(route="/callback", on_load=SsoState.callback)
+def callback() -> rx.Component:
+    return rx.container()
+
+
+@rx.page(route="/logout", on_load=SsoState.logout)
+def logout() -> rx.Component:
+    return rx.container("Logged out")
+
+
+@rx.page(route="/", on_load=SsoState.require_auth)
+def home() -> rx.Component:
+    return rx.container(rx.cond(SsoState.check_auth, auth_view(), unauth_view()))
+
+
+def auth_view() -> rx.Component:
+    return rx.vstack(
+        rx.text(f"Hello {SsoState.token['name']}"),
+        rx.button("logout", on_click=SsoState.logout),
+    )
+
+
+def unauth_view() -> rx.Component:
+    return rx.text("Unauthorized, redirected...")

azure-auth/requirements.txt

@@ -0,0 +1,2 @@
+reflex==0.5.10
+msal==1.31.0

azure-auth/rxconfig.py

@@ -0,0 +1,5 @@
+import reflex as rx
+
+config = rx.Config(
+    app_name="azure_auth",
+)