Skip to content

Auth Integration

Below is a reference‑style recipe you can drop into the Cookbook for Auth integration using Auth0. It mirrors the Gmail OAuth recipe: Authorization Code + PKCE, Jetty HttpClient via a thread‑safe singleton, session storage for tokens, and a sample call to /userinfo (or any API with an Audience).

🔐 Recipe: Auth0 OAuth 2.0 (Authorization Code + PKCE)

SØAD supports browser OAuth flows. With Auth0, use Authorization Code with PKCE:

  1. Redirect to Auth0’s /authorize (include code_challenge)
  2. Handle your callback, validate state, exchange code at /oauth/token with code_verifier
  3. Store access_token (short‑lived) and optionally refresh_token (if enabled)
  4. Call /userinfo or your protected API with Authorization: Bearer <access_token>
  5. Refresh tokens if configured and returned

Auth0 settings (Dashboard → Applications → Your App):

  • Application Type: Regular Web Application
  • Allowed Callback URLs: https://<host><ctxPath>/t/auth/auth0/callback
  • Allowed Logout URLs (optional): https://<host>
  • Allowed Web Origins: https://<host>
  • Token Endpoint Auth Method: choose accordingly; with PKCE, client_secret is typically not needed.
  • Audience (optional): your API identifier if you want an API access token (not only OIDC userinfo).
  • Scopes: e.g. openid profile email (+ API scopes if using Audience)

🛠 Utility (shared): http_client_provider.py (singleton)

from org.eclipse.jetty.client import HttpClient
from org.eclipse.jetty.util.ssl import SslContextFactory
from java.util.concurrent.locks import ReentrantLock

class HttpClientProvider:
    _client = None
    _lock = ReentrantLock()

    @classmethod
    def get_client(cls):
        if cls._client is None:
            cls._lock.lock()
            try:
                if cls._client is None:
                    ssl = SslContextFactory.Client()
                    client = HttpClient(ssl)
                    client.start()
                    cls._client = client
            finally:
                cls._lock.unlock()
        return cls._client

🧩 Transaction: auth0.py

from utils import render, Log
from http_client_provider import HttpClientProvider

from java.util import Base64, UUID
from java.net import URLEncoder
from java.nio.charset import StandardCharsets
from java.security import SecureRandom, MessageDigest
from org.eclipse.jetty.http import HttpHeader
from org.eclipse.jetty.client.util import StringContentProvider
from java.lang import System
import json

# ---- Configure via environment or sufia.properties (preferred) ----
AUTH0_DOMAIN  = System.getenv("AUTH0_DOMAIN")  or "your-tenant.eu.auth0.com"
CLIENT_ID     = System.getenv("AUTH0_CLIENT_ID") or "YOUR_CLIENT_ID"
AUDIENCE      = System.getenv("AUTH0_AUDIENCE")  # e.g., "https://api.yourapp.com" (optional)
SCOPES        = System.getenv("AUTH0_SCOPES")    or "openid profile email"
# --------------------------------------------------------------------

AUTH_URL  = "https://%s/authorize"   % AUTH0_DOMAIN
TOKEN_URL = "https://%s/oauth/token" % AUTH0_DOMAIN
USERINFO  = "https://%s/userinfo"    % AUTH0_DOMAIN

class Auth0(Layout):

    # ----- Helpers (PKCE, url encoding) -----
    def _base64url(self, bts):
        return Base64.getUrlEncoder().withoutPadding().encodeToString(bts)

    def _random_bytes(self, n=32):
        rnd = SecureRandom()
        buf = bytearray(n)
        rnd.nextBytes(buf)
        return bytes(buf)

    def _code_verifier(self):
        return self._base64url(self._random_bytes(64))

    def _code_challenge(self, verifier):
        digest = MessageDigest.getInstance("SHA-256").digest(verifier.encode("UTF-8"))
        return self._base64url(digest)

    def _q(self, k, v):
        return "%s=%s" % (k, URLEncoder.encode(v, StandardCharsets.UTF_8.name()))

    # ----- Step 1: Redirect to Auth0 with PKCE -----
    def login(self, ctx):
        session = ctx.getSession()
        verifier  = self._code_verifier()
        challenge = self._code_challenge(verifier)
        state     = UUID.randomUUID().toString()

        session.setAttribute("auth0_pkce_verifier", verifier)
        session.setAttribute("auth0_oauth_state", state)

        redirect_uri = "%s/t/auth/auth0/callback" % ctx.ctxPath

        params = [
            self._q("client_id", CLIENT_ID),
            self._q("redirect_uri", redirect_uri),
            self._q("response_type", "code"),
            self._q("scope", SCOPES),
            self._q("state", state),
            self._q("code_challenge", challenge),
            self._q("code_challenge_method", "S256")
        ]
        if AUDIENCE:
            params.append(self._q("audience", AUDIENCE))

        auth_url = AUTH_URL + "?" + "&".join(params)
        ctx.redirect(auth_url)

    # ----- Step 2: Callback & code exchange -----
    def callback(self, ctx):
        req     = ctx.getRequest()
        session = ctx.getSession()

        code  = req.getParameter("code")
        state = req.getParameter("state")
        err   = req.getParameter("error")

        if err:
            ctx.output["error"] = "Authorization error: %s" % err
            ctx.go_to = render.as_view(ctx, "auth_result")
            return

        if not state or state != session.getAttribute("auth0_oauth_state"):
            ctx.output["error"] = "Invalid state"
            ctx.go_to = render.as_view(ctx, "auth_result")
            return

        verifier = session.getAttribute("auth0_pkce_verifier")
        if not verifier:
            ctx.output["error"] = "Missing PKCE verifier"
            ctx.go_to = render.as_view(ctx, "auth_result")
            return

        redirect_uri = "%s/t/auth/auth0/callback" % ctx.ctxPath

        form = [
            self._q("grant_type", "authorization_code"),
            self._q("client_id", CLIENT_ID),
            self._q("code_verifier", verifier),
            self._q("code", code),
            self._q("redirect_uri", redirect_uri)
        ]
        body = "&".join(form)

        try:
            client = HttpClientProvider.get_client()
            reqx = client.POST(TOKEN_URL)
            reqx.header(HttpHeader.CONTENT_TYPE, "application/x-www-form-urlencoded")
            reqx.content(StringContentProvider(body, "utf-8"))

            res = reqx.send()
            text = res.getContentAsString()

            tok = json.loads(text)
            access_token  = tok.get("access_token")
            refresh_token = tok.get("refresh_token")  # returned only if enabled
            id_token      = tok.get("id_token")

            if not access_token:
                ctx.output["error"] = "Token exchange failed"
                ctx.output["response"] = text
                ctx.go_to = render.as_view(ctx, "auth_result")
                return

            session.setAttribute("auth0_access_token", access_token)
            if refresh_token:
                session.setAttribute("auth0_refresh_token", refresh_token)
            if id_token:
                session.setAttribute("auth0_id_token", id_token)

            ctx.output["message"] = "Auth0 login successful."
            ctx.go_to = render.as_view(ctx, "auth_result")

        except Exception as e:
            Log.error("Auth0 token exchange error: %s" % str(e))
            ctx.output["error"] = "Auth0 token exchange error."
            ctx.go_to = render.as_view(ctx, "auth_result")

    # ----- Step 3: Call /userinfo (or your API) -----
    def userinfo(self, ctx):
        session = ctx.getSession()
        token = session.getAttribute("auth0_access_token")
        if not token:
            ctx.redirect("%s/t/auth/auth0/login" % ctx.ctxPath)
            return

        try:
            client = HttpClientProvider.get_client()
            req = client.newRequest(USERINFO)
            req.header(HttpHeader.AUTHORIZATION, "Bearer %s" % token)
            res = req.send()

            ctx.output["status"] = res.getStatus()
            ctx.output["result"] = res.getContentAsString()

        except Exception as e:
            Log.error("Auth0 /userinfo call failed: %s" % str(e))
            ctx.output["error"] = "Auth0 API call failed."

        ctx.go_to = render.as_view(ctx, "auth0_result")

📄 Views

_auth0/auth_result.html

<h2>Auth0 Result</h2>
{{#if error}}
  <div class="alert alert-danger">Error: {{error}}</div>
  {{#if response}}<pre>{{response}}</pre>{{/if}}
{{else}}
  <div class="alert alert-success">{{message}}</div>
  <p><a href="{{ctxPath}}/t/auth/auth0/userinfo">View /userinfo</a></p>
{{/if}}

_auth0/auth0_result.html

<h2>Auth0 /userinfo (Raw Response)</h2>
{{#if error}}
  <div class="alert alert-danger">{{error}}</div>
{{else}}
  <div class="alert alert-success"><strong>Status:</strong> {{status}}</div>
  <pre>{{result}}</pre>
{{/if}}

🔎 Notes & Options

  • Audience: Supply AUDIENCE to obtain an API access token for your backend; otherwise token is typically for OIDC /userinfo.
  • Refresh Tokens: Enable in Auth0 application & pass offline_access scope to receive refresh_token.
  • Logout: You can implement logout by clearing session and optionally calling https://<domain>/v2/logout?client_id=...&returnTo=....
  • Storage: Session is fine for demos. For production, store tokens securely (encrypted DB, vault, KMS).
  • Scopes: Minimize to what you need (openid profile email + API scopes).
  • Callback URL: Must match Allowed Callback URLs in Auth0 app settings exactly.

This pattern generalizes to any OAuth provider: adjust issuer domain, endpoints, scopes, and audience as required.