OAuth 2.0 - Tutorial 3 - Expose and Access a Web API - Resource Server

cd fastapi_app

uv init
uv add fastapi uvicorn[standard] pydantic pwdlib[argon2] python-multipart python-jose
# Activate the virtual environment with those dependencies installed:
source .venv/bin/activate

# Type hints and annotation support
from typing import Annotated

# FastAPI and related imports
from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import OAuth2AuthorizationCodeBearer, SecurityScopes
# JWT decoding
from jose import jwt
# Data validation
from pydantic import BaseModel


# Entra ID config
TENANT_ID = "<ADD-YOUR-ENTRA-ID-TENANT-ID>" #TODO
OPENID_CONFIG_URL = f"https://login.microsoftonline.com/{TENANT_ID}/v2.0/.well-known/openid-configuration"
API_AUDIENCE = "<ADD-YOUR-API-AUDIENCE>" # First Expose Web API to get your API Audience #TODO
ALGORITHM = "RS256"


import requests

# Fetch JWKS URL from OpenID configuration at startup
# This gets the public keys used to verify JWTs
resp = requests.get(OPENID_CONFIG_URL)
resp.raise_for_status()
JWKS_URL = resp.json()["jwks_uri"]

# Get the JSON Web Key Set (JWKS) from Entra ID
def get_jwks():
    resp = requests.get(JWKS_URL)
    resp.raise_for_status()
    return resp.json()["keys"]

# Find the public key in JWKS that matches the JWT's key ID (kid)
def get_public_key(token):
    jwks = get_jwks()
    unverified_header = jwt.get_unverified_header(token)
    for key in jwks:
        if key["kid"] == unverified_header["kid"]:
            return key
    raise Exception("Public key not found.")

# Pydantic model for token data
class TokenData(BaseModel):
    username: str | None = None  # Username from token (not used currently)
    scopes: list[str] = []  # List of scopes from token

# OAuth2 scheme for FastAPI security
oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/authorize",  # Entra ID auth URL
    tokenUrl=f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token",  # Entra ID token URL
    scopes={
        "data.read": "Read protected data.",
        "data.write": "Write protected data.",
    },
)

# Create FastAPI app
app = FastAPI()

# Dependency to get the current user from the JWT token
# Checks for required scopes and validates the token
async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    # Build the authenticate header value for WWW-Authenticate
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    # Exception to raise if authentication fails (invalid, missing, or expired token)
    # This is required for proper OAuth2 error signaling, regardless of grant type.
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        # Get public key and decode JWT
        public_key = get_public_key(token)
        payload = jwt.decode(
            token,
            public_key,
            algorithms=[ALGORITHM],
            audience=API_AUDIENCE,
            options={"verify_exp": True, "verify_aud": True},
        )
        # Extract scopes from token payload
        scope: str = payload.get("scp", "") or payload.get("scope", "")
        token_scopes = scope.split(" ")
        token_data = TokenData(scopes=token_scopes)
    except Exception:
        # Token is invalid or cannot be decoded
        raise credentials_exception
    # Check if all required scopes are present in the token
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return token_data



# Protected endpoint: requires 'data.read' scope
@app.get("/data/read")
async def read_data(
    current_user: Annotated[
        TokenData, Security(get_current_user, scopes=["data.read"])
    ],
):
    """Return protected data if user has 'data.read' scope."""
    return {
        "data": "This is protected data you can read.",
        "current-user-scopes": current_user.scopes,
    }

# Protected endpoint: requires 'data.write' scope
@app.post("/data/write")
async def write_data(
    current_user: Annotated[
        TokenData, Security(get_current_user, scopes=["data.write"])
    ],
):
    """Allow writing protected data if user has 'data.write' scope."""
    return {"result": "You wrote protected data!", "current-user-scopes": current_user.scopes}

from flask import Flask, session, redirect, request, url_for
from flask_session import Session
from dotenv import load_dotenv
import msal
import uuid
import os
import requests

app = Flask(__name__)
app.secret_key = os.urandom(24)  # Needed for session
# Configure Flask-Session to use filesystem (server-side session)
app.config["SESSION_TYPE"] = "filesystem"
app.config["SESSION_PERMANENT"] = False
app.config["SESSION_USE_SIGNER"] = True
Session(app)
# Specify the path to the .env file
dotenv_path = ".env"  # Update this with the actual path

# Load the .env file
load_dotenv(dotenv_path=dotenv_path)

# Config
config = {
    "client_id": os.getenv("CLIENT_ID"),
    "authority": f"https://login.microsoftonline.com/{os.getenv('TENANT_ID')}",
    # Add FastAPI API scopes for read and write
    "scope": [
        "User.Read",
        #TODO - replace with your api scopes
        "api://c750c37c-6c67-4211-a9e2-6f035a897cfa/data.read", 
        "api://c750c37c-6c67-4211-a9e2-6f035a897cfa/data.write",
    ],
    "redirect_uri": os.getenv("REDIRECT_URI"),
    "client_secret": os.getenv("CLIENT_SECRET"),
}


def build_msal_app():
    return msal.ConfidentialClientApplication(
        client_id=config["client_id"],
        authority=config["authority"],
        client_credential=config["client_secret"],
    )

# On the index page, add simple links to our backend FastAPI endpoints
@app.route("/")
def index():
    if "user" in session:
        user = session["user"]
        return f"""
        <h2>Welcome {user['displayName']}</h2>
        <p>Email: {user.get('mail') or user.get('userPrincipalName')}</p>
    <a href="/logout">Logout</a><br>
    <a href="/api/read">Read Protected Data (FastAPI)</a><br>
    <a href="/api/write">Write Protected Data (FastAPI)</a>
        """
    return '<a href="/login">Login with Microsoft</a>'

@app.route("/api/read")
def api_read():
    api_access_token = session.get("api_access_token")
    if not api_access_token:
        return redirect(url_for("login"))
    # Call the FastAPI backend read endpoint
    api_url = "http://localhost:8000/data/read"
    resp = requests.get(api_url, headers={"Authorization": f"Bearer {api_access_token}"})
    if resp.ok:
        return f"<pre>{resp.json()}</pre><a href='/'>Back</a>"
    return f"Error calling API: {resp.status_code} {resp.text} <a href='/'>Back</a>", 500


@app.route("/api/write", methods=["GET", "POST"])
def api_write():
    api_access_token = session.get("api_access_token")
    if not api_access_token:
        return redirect(url_for("login"))
    # Call the FastAPI backend write endpoint
    api_url = "http://localhost:8000/data/write"
    resp = requests.post(api_url, headers={"Authorization": f"Bearer {api_access_token}"})
    if resp.ok:
        return f"<pre>{resp.json()}</pre><a href='/'>Back</a>"
    return f"Error calling API: {resp.status_code} {resp.text} <a href='/'>Back</a>", 500


@app.route("/login")
def login():
    msal_app = build_msal_app()
    flow = msal_app.initiate_auth_code_flow(
        scopes=config["scope"],
        redirect_uri=config["redirect_uri"],
        state=str(uuid.uuid4()),
    )
    session["auth_flow"] = flow
    return redirect(flow["auth_uri"])


@app.route("/callback")
def callback():
    msal_app = build_msal_app()
    try:
        result = msal_app.acquire_token_by_auth_code_flow(
            session.get("auth_flow", {}), request.args
        )
    except ValueError:
        return "Authentication failed", 400

    if "access_token" in result:
        session["access_token"] = result["access_token"]  # This is likely a Graph token

        # Call Microsoft Graph to get user info
        graph_resp = requests.get(
            "https://graph.microsoft.com/v1.0/me",
            headers={"Authorization": f"Bearer {result['access_token']}"},
        )
        if graph_resp.ok:
            session["user"] = graph_resp.json()
        else:
            return f"Graph API error: {graph_resp.text}", 500

        # Now acquire a token for the FastAPI API (audience = API)
        account = msal_app.get_accounts()[0] if msal_app.get_accounts() else None
        api_scopes = [
            #TODO - replace with your api scopes
            "api://c750c37c-6c67-4211-a9e2-6f035a897cfa/data.read",
            "api://c750c37c-6c67-4211-a9e2-6f035a897cfa/data.write",
        ]
        api_token_result = msal_app.acquire_token_silent(api_scopes, account=account)
        if not api_token_result:
            # If silent fails, use refresh token if available
            api_token_result = msal_app.acquire_token_by_authorization_code(
                request.args.get("code"),
                scopes=api_scopes,
                redirect_uri=config["redirect_uri"],
            )
        if "access_token" in api_token_result:
            session["api_access_token"] = api_token_result["access_token"]
        else:
            return f"API token error: {api_token_result.get('error_description')}", 400

        return redirect(url_for("index"))

    return f"Error: {result.get('error_description')}", 400


@app.route("/logout")
def logout():
    session.clear()
    return redirect(url_for("index"))


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001, debug=True)

uvicorn main:app --host=0.0.0.0 --port=8000

flask --app main.py run --host=0.0.0.0 --port=5001