MSAL Based Authentication - Redirect URL error

[dharanidharan59]

@victordibia

I am working on the implementation of MSAL Based authentication in AGS.
Here's my code for the authentication:

`import json
import secrets
import sys
from abc import ABC, abstractmethod
from urllib.parse import urlencode

import httpx
from loguru import logger

# Configure logger to print to console
logger.remove()  # Remove default handlers
logger.add(sys.stdout, level="DEBUG", colorize=True)

from .exceptions import ConfigurationException, ProviderAuthException
from .models import AuthConfig, GithubAuthConfig, User


class AuthProvider(ABC):
    """Base authentication provider interface."""

    @abstractmethod
    async def get_login_url(self) -> str:
        """Return the URL for initiating login."""
        pass

    @abstractmethod
    async def process_callback(self, code: str, state: str | None = None) -> User:
        """Process the OAuth callback code and return user data."""
        pass

    @abstractmethod
    async def validate_token(self, token: str) -> bool:
        """Validate a provider token and return boolean indicating validity."""
        pass


class NoAuthProvider(AuthProvider):
    """Default provider that always authenticates (for development)."""

    def __init__(self):
        self.default_user = User(
            id="[email protected]", name="Default User", email="[email protected]", provider="none"
        )

    async def get_login_url(self) -> str:
        """Return the URL for initiating login."""
        return "/api/auth/callback?automatic=true"

    async def process_callback(self, code: str | None = None, state: str | None = None) -> User:
        """Process the OAuth callback code and return user data."""
        return self.default_user

    async def validate_token(self, token: str) -> bool:
        """Validate a provider token and return boolean indicating validity."""
        return True


class GithubAuthProvider(AuthProvider):
    """GitHub OAuth authentication provider."""

    def __init__(self, config: AuthConfig):
        print("Initializing GithubAuthProvider")
        if not config.github:
            raise ConfigurationException("GitHub auth configuration is missing")

        self.config = config.github
        self.client_id = self.config.client_id
        self.client_secret = self.config.client_secret
        self.callback_url = self.config.callback_url
        self.scopes = self.config.scopes
        print(f"GitHub Provider initialized with client_id: {self.client_id}")
        print(f"GitHub callback URL: {self.callback_url}")

    async def get_login_url(self) -> str:
        """Return the GitHub OAuth login URL."""
        print("GithubAuthProvider.get_login_url() called")
        state = secrets.token_urlsafe(32)  # Generate a secure random state
        params = {
            "client_id": self.client_id,
            "redirect_uri": self.callback_url,
            "scope": " ".join(self.scopes),
            "state": state,
            "allow_signup": "true",
        }
        url = f"https://github.com/login/oauth/authorize?{urlencode(params)}"
        print(f"GitHub login URL generated: {url}")
        return url

    async def process_callback(self, code: str, state: str | None = None) -> User:
        """Exchange code for access token and get user info."""
        print(f"GithubAuthProvider.process_callback() called with code: {code[:5]}...")
        if not code:
            raise ProviderAuthException("github", "Authorization code is missing")

        # Exchange code for access token
        token_url = "https://github.com/login/oauth/access_token"
        token_data = {
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "code": code,
            "redirect_uri": self.callback_url,
        }

        print("Exchanging GitHub code for access token")
        async with httpx.AsyncClient() as client:
            token_response = await client.post(token_url, data=token_data, headers={"Accept": "application/json"})

            if token_response.status_code != 200:
                logger.error(f"GitHub token exchange failed: {token_response.text}")
                print(f"GitHub token exchange failed: {token_response.text}")
                raise ProviderAuthException("github", "Failed to exchange code for access token")

            token_json = token_response.json()
            access_token = token_json.get("access_token")

            if not access_token:
                logger.error(f"No access token in GitHub response: {token_json}")
                print(f"No access token in GitHub response: {token_json}")
                raise ProviderAuthException("github", "No access token received")

            # Get user info with the access token
            print("Getting GitHub user info with access token")
            user_response = await client.get(
                "https://api.github.com/user",
                headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
            )

            if user_response.status_code != 200:
                logger.error(f"GitHub user info fetch failed: {user_response.text}")
                print(f"GitHub user info fetch failed: {user_response.text}")
                raise ProviderAuthException("github", "Failed to fetch user information")

            user_data = user_response.json()
            print(f"GitHub user data received: {user_data.get('login')}")

            # Get user emails if scope includes email
            email = None
            if "user:email" in self.scopes:
                print("Getting GitHub user emails")
                email_response = await client.get(
                    "https://api.github.com/user/emails",
                    headers={"Authorization": f"token {access_token}", "Accept": "application/json"},
                )

                if email_response.status_code == 200:
                    emails = email_response.json()
                    primary_emails = [e for e in emails if e.get("primary") is True]
                    if primary_emails:
                        email = primary_emails[0].get("email")
                        print(f"Found primary email: {email}")

            # Create User object
            user = User(
                id=str(user_data.get("id")),
                name=user_data.get("name") or user_data.get("login"),
                email=email,
                avatar_url=user_data.get("avatar_url"),
                provider="github",
                metadata={
                    "login": user_data.get("login"),
                    "github_id": user_data.get("id"),
                    "access_token": access_token,
                },
            )
            print(f"Returning GitHub user: {user.name} (ID: {user.id})")
            return user

    async def validate_token(self, token: str) -> bool:
        """Validate a GitHub access token."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.github.com/user", headers={"Authorization": f"token {token}", "Accept": "application/json"}
            )
            return response.status_code == 200


class MSALAuthProvider(AuthProvider):
    """Microsoft Authentication Library (MSAL) provider."""

    def __init__(self, config: AuthConfig):
        logger.debug("\n" + "="*50)
        logger.debug("MSAL PROVIDER INITIALIZATION")
        logger.debug("="*50)
        
        if not config.msal:
            logger.error("MSAL auth configuration is missing")
            raise ConfigurationException("MSAL auth configuration is missing")

        self.config = config.msal
        self.client_id = self.config.client_id
        self.client_secret = self.config.client_secret
        self.callback_url = self.config.callback_url
        self.tenant_id = self.config.tenant_id
        self.scopes = self.config.scopes

        logger.debug(f"MSAL Provider Configuration:")
        logger.debug(f"Client ID: {self.client_id}")
        logger.debug(f"Tenant ID: {self.tenant_id}")
        logger.debug(f"Callback URL: {self.callback_url}")
        logger.debug(f"Scopes: {self.scopes}")
        logger.debug("="*50 + "\n")

    async def get_login_url(self) -> str:
        logger.debug("\n" + "="*50)
        logger.debug("GENERATING MSAL LOGIN URL")
        logger.debug("="*50)
        
        state = secrets.token_urlsafe(32)
        params = {
            "client_id": self.client_id,
            "response_type": "code",
            "redirect_uri": self.callback_url,
            "scope": " ".join(self.scopes),
            "state": state,
            "response_mode": "query"
        }
        
        login_url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/authorize?{urlencode(params)}"
        
        logger.debug(f"Generated URL: {login_url}")
        logger.debug("="*50 + "\n")
        
        return login_url

    async def process_callback(self, code: str, state: str | None = None) -> User:
        """Process the MSAL callback."""
        # Placeholder - would use MSAL library to process code and get token/user info
        return User(id="msal_user_id", name="MSAL User", provider="msal")

    async def validate_token(self, token: str) -> bool:
        """Validate an MSAL token."""
        # Placeholder - would validate token with MSAL library
        return False


class FirebaseAuthProvider(AuthProvider):
    """Firebase authentication provider."""

    def __init__(self, config: AuthConfig):
        if not config.firebase:
            raise ConfigurationException("Firebase auth configuration is missing")

        self.config = config.firebase
        # Firebase provider implementation would go here
        # This is a placeholder - full implementation would use Firebase Admin SDK

    async def get_login_url(self) -> str:
        """Return information for Firebase auth (used differently than OAuth)."""
        # Firebase auth is typically handled on the client side
        return json.dumps(
            {"apiKey": self.config.api_key, "authDomain": self.config.auth_domain, "projectId": self.config.project_id}
        )

    async def process_callback(self, code: str, state: str | None = None) -> User:
        """Process a Firebase ID token."""
        # Placeholder - would verify Firebase ID token and get user info
        return User(id="firebase_user_id", name="Firebase User", provider="firebase")

    async def validate_token(self, token: str) -> bool:
        """Validate a Firebase ID token."""
        # Placeholder - would validate token with Firebase Admin SDK
        return False
`

Also I am passing the auth parameters using auth.yaml file like below

`
type: msal
jwt_secret: "xxx"
token_expiry_minutes: 60
msal:
  tenant_id:   # Your Azure AD tenant ID
  client_id:  # Your registered app's client ID
  client_secret: # Your registered app's client secret
  callback_url: "http://localhost:8000"
  scopes: ["User.Read"]
`

However the login URL is not getting constructed and giving the below error.
Not able to identify from where the placeholder url is taken. (This URL( https://login.microsoftonline.com/placeholder ) was there previously without implementation).
Tried clearing cache, but still the same issue.

Could you please help here

[victordibia]

Hi,

We currently support just GH Auth (experimental).
The error above also seems to be related to your MSAL implementation and not Autogen Studio specific? Maybe get help on the MSAL GH?

[dharanidharan59]

Hi @victordibia
Thanks for your reply

I'm curious when will the MSAL login functionality would be shipped as a feature!!!.
Please let us know
It will be more useful for our usecase