Okta Testing Strategies: SAML Flows, Okta Workflows, and SDK Mocking

# Environment variables for test configuration
OKTA_ORG_URL=https://dev-12345.oktapreview.com
OKTA_API_TOKEN=00xyzABCDEFG...       <span class="hljs-comment"># From Okta Admin > Security > API > Tokens
OKTA_CLIENT_ID=0oa1abc2defg3hij4kl5
OKTA_CLIENT_SECRET=xyz-secret
OKTA_ISSUER=https://dev-12345.oktapreview.com/oauth2/default

// src/auth/okta-middleware.js
import { OktaJwtVerifier } from '@okta/jwt-verifier';

export async function verifyToken(req, res, next) {
  const authHeader = req.headers.authorization;
  if (!authHeader?.startsWith('Bearer ')) {
    return res.status(401).json({ error: 'No token provided' });
  }
  
  const token = authHeader.split(' ')[1];
  
  try {
    const verifier = new OktaJwtVerifier({
      issuer: process.env.OKTA_ISSUER,
      clientId: process.env.OKTA_CLIENT_ID
    });
    
    const jwt = await verifier.verifyAccessToken(token, 'api://default');
    req.user = { id: jwt.claims.sub, email: jwt.claims.email, groups: jwt.claims.groups };
    next();
  } catch (err) {
    return res.status(401).json({ error: 'Invalid token', details: err.message });
  }
}

// tests/unit/okta-middleware.test.js
import { jest } from '@jest/globals';
import { verifyToken } from '../../src/auth/okta-middleware.js';

// Mock the Okta JWT verifier
jest.mock('@okta/jwt-verifier', () => ({
  OktaJwtVerifier: jest.fn().mockImplementation(() => ({
    verifyAccessToken: jest.fn()
  }))
}));

import { OktaJwtVerifier } from '@okta/jwt-verifier';

describe('verifyToken middleware', () => {
  let mockReq, mockRes, mockNext;
  let mockVerifyAccessToken;

  beforeEach(() => {
    mockReq = { headers: {} };
    mockRes = {
      status: jest.fn().mockReturnThis(),
      json: jest.fn().mockReturnThis()
    };
    mockNext = jest.fn();
    
    mockVerifyAccessToken = jest.fn();
    OktaJwtVerifier.mockImplementation(() => ({
      verifyAccessToken: mockVerifyAccessToken
    }));
  });

  it('returns 401 when no Authorization header provided', async () => {
    await verifyToken(mockReq, mockRes, mockNext);
    
    expect(mockRes.status).toHaveBeenCalledWith(401);
    expect(mockNext).not.toHaveBeenCalled();
  });

  it('returns 401 when Authorization is not Bearer', async () => {
    mockReq.headers.authorization = 'Basic dXNlcjpwYXNz';
    
    await verifyToken(mockReq, mockRes, mockNext);
    
    expect(mockRes.status).toHaveBeenCalledWith(401);
    expect(mockNext).not.toHaveBeenCalled();
  });

  it('calls next() with user data on valid token', async () => {
    mockReq.headers.authorization = 'Bearer valid-token';
    mockVerifyAccessToken.mockResolvedValue({
      claims: {
        sub: 'user-123',
        email: 'user@example.com',
        groups: ['admins', 'developers']
      }
    });
    
    await verifyToken(mockReq, mockRes, mockNext);
    
    expect(mockNext).toHaveBeenCalled();
    expect(mockReq.user).toEqual({
      id: 'user-123',
      email: 'user@example.com',
      groups: ['admins', 'developers']
    });
  });

  it('returns 401 on expired token', async () => {
    mockReq.headers.authorization = 'Bearer expired-token';
    mockVerifyAccessToken.mockRejectedValue(new Error('Token expired'));
    
    await verifyToken(mockReq, mockRes, mockNext);
    
    expect(mockRes.status).toHaveBeenCalledWith(401);
    expect(mockRes.json).toHaveBeenCalledWith(
      expect.objectContaining({ error: 'Invalid token' })
    );
  });

  it('returns 401 on wrong audience', async () => {
    mockReq.headers.authorization = 'Bearer wrong-audience-token';
    mockVerifyAccessToken.mockRejectedValue(
      new Error("Jwt audience claim 'api://wrong' does not match 'api://default'")
    );
    
    await verifyToken(mockReq, mockRes, mockNext);
    
    expect(mockRes.status).toHaveBeenCalledWith(401);
  });
});

# tests/unit/test_okta_auth.py
from unittest.mock import MagicMock, patch
import pytest
from app import create_app

@pytest.fixture
def app():
    app = create_app(testing=True)
    return app

@pytest.fixture
def client(app):
    return app.test_client()

def make_mock_token_payload(sub="user-123", groups=None, expired=False):
    import time
    return {
        "sub": sub,
        "email": f"{sub}@example.com",
        "groups": groups or ["Everyone"],
        "iat": int(time.time()),
        "exp": int(time.time()) + (-3600 if expired else 3600),
        "iss": "https://dev-12345.oktapreview.com/oauth2/default",
        "aud": "api://default"
    }

@patch("app.auth.jwt.decode")
def test_protected_route_with_valid_token(mock_decode, client):
    mock_decode.return_value = make_mock_token_payload()
    
    response = client.get(
        "/api/profile",
        headers={"Authorization": "Bearer fake-valid-token"}
    )
    assert response.status_code == 200
    assert response.json["user"]["email"] == "user-123@example.com"

@patch("app.auth.jwt.decode")
def test_protected_route_blocked_without_token(mock_decode, client):
    response = client.get("/api/profile")
    assert response.status_code == 401
    mock_decode.assert_not_called()

@patch("app.auth.jwt.decode")
def test_admin_route_requires_admin_group(mock_decode, client):
    # User not in admin group
    mock_decode.return_value = make_mock_token_payload(groups=["Everyone"])
    response = client.get(
        "/api/admin/users",
        headers={"Authorization": "Bearer user-token"}
    )
    assert response.status_code == 403

@patch("app.auth.jwt.decode")
def test_admin_route_allows_admin_group(mock_decode, client):
    mock_decode.return_value = make_mock_token_payload(groups=["Everyone", "Admins"])
    response = client.get(
        "/api/admin/users",
        headers={"Authorization": "Bearer admin-token"}
    )
    assert response.status_code == 200

# tests/integration/test_okta_oidc.py
import pytest
import httpx
import os

OKTA_ORG = os.environ["OKTA_ORG_URL"]
CLIENT_ID = os.environ["OKTA_CLIENT_ID"]
CLIENT_SECRET = os.environ["OKTA_CLIENT_SECRET"]
# Test users pre-created in Okta preview org
TEST_USER_EMAIL = os.environ["OKTA_TEST_USER_EMAIL"]
TEST_USER_PASSWORD = os.environ["OKTA_TEST_USER_PASSWORD"]

def get_access_token(username, password):
    """Get token via Resource Owner Password grant (preview orgs only)."""
    r = httpx.post(
        f"{OKTA_ORG}/oauth2/default/v1/token",
        data={
            "grant_type": "password",
            "username": username,
            "password": password,
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "scope": "openid profile email"
        }
    )
    assert r.status_code == 200, f"Token request failed: {r.text}"
    return r.json()

def test_valid_credentials_return_access_token():
    tokens = get_access_token(TEST_USER_EMAIL, TEST_USER_PASSWORD)
    
    assert "access_token" in tokens
    assert "id_token" in tokens
    assert tokens["token_type"] == "Bearer"
    assert tokens["expires_in"] > 0

def test_invalid_credentials_rejected():
    r = httpx.post(
        f"{OKTA_ORG}/oauth2/default/v1/token",
        data={
            "grant_type": "password",
            "username": TEST_USER_EMAIL,
            "password": "wrong-password",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "scope": "openid"
        }
    )
    assert r.status_code == 401 or r.status_code == 400
    assert "error" in r.json()

def test_id_token_contains_user_claims():
    import jwt
    tokens = get_access_token(TEST_USER_EMAIL, TEST_USER_PASSWORD)
    
    # Decode without verification for claim inspection (verification tested separately)
    payload = jwt.decode(
        tokens["id_token"],
        options={"verify_signature": False}
    )
    
    assert payload["sub"] is not None
    assert payload["email"] == TEST_USER_EMAIL
    assert payload["iss"] == f"{OKTA_ORG}/oauth2/default"

def test_access_token_introspection():
    tokens = get_access_token(TEST_USER_EMAIL, TEST_USER_PASSWORD)
    
    r = httpx.post(
        f"{OKTA_ORG}/oauth2/default/v1/introspect",
        data={
            "token": tokens["access_token"],
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET
        }
    )
    assert r.status_code == 200
    info = r.json()
    assert info["active"] is True
    assert info["username"] == TEST_USER_EMAIL

def test_token_revocation():
    tokens = get_access_token(TEST_USER_EMAIL, TEST_USER_PASSWORD)
    
    # Revoke the token
    r = httpx.post(
        f"{OKTA_ORG}/oauth2/default/v1/revoke",
        data={
            "token": tokens["access_token"],
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET
        }
    )
    assert r.status_code == 200
    
    # Introspect — should show as inactive
    introspect = httpx.post(
        f"{OKTA_ORG}/oauth2/default/v1/introspect",
        data={
            "token": tokens["access_token"],
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET
        }
    )
    assert introspect.json()["active"] is False

# tests/e2e/test_saml_sso.py
import pytest
from playwright.sync_api import sync_playwright, Page

OKTA_ORG = "https://dev-12345.oktapreview.com"
APP_URL = "http://localhost:3000"

@pytest.fixture(scope="session")
def browser():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        yield browser
        browser.close()

@pytest.fixture
def page(browser):
    page = browser.new_page()
    yield page
    page.close()

def test_saml_sso_redirects_to_okta(page: Page):
    """Clicking 'Sign in with SSO' redirects to Okta login."""
    page.goto(f"{APP_URL}/login")
    page.click("text=Sign in with SSO")
    
    # Should redirect to Okta
    page.wait_for_url(f"{OKTA_ORG}/**", timeout=10000)
    assert OKTA_ORG in page.url

def test_saml_sso_full_flow(page: Page):
    """Complete SAML SSO login with test credentials."""
    # Navigate to protected resource (triggers SSO)
    page.goto(f"{APP_URL}/dashboard")
    
    # Should redirect to Okta
    page.wait_for_url(f"{OKTA_ORG}/**", timeout=10000)
    
    # Fill Okta login form
    page.fill("#okta-signin-username", "testuser@example.com")
    page.fill("#okta-signin-password", "TestPassword123!")
    page.click("#okta-signin-submit")
    
    # Should redirect back to the app
    page.wait_for_url(f"{APP_URL}/**", timeout=15000)
    
    # Should be on the dashboard now
    assert "/dashboard" in page.url
    page.wait_for_selector("h1:has-text('Dashboard')", timeout=5000)

def test_saml_sso_logout(page: Page):
    """Logging out terminates both app session and Okta session."""
    # First, log in
    test_saml_sso_full_flow(page)
    
    # Click logout
    page.click("[data-testid='logout-button']")
    
    # Should redirect to login page or Okta logout
    page.wait_for_url(f"{APP_URL}/login", timeout=10000)
    
    # Verify session is terminated — visiting protected page redirects to login
    page.goto(f"{APP_URL}/dashboard")
    assert "/login" in page.url or OKTA_ORG in page.url

# tests/integration/test_okta_workflows.py
import httpx
import os
import time

OKTA_ORG = os.environ["OKTA_ORG_URL"]
API_TOKEN = os.environ["OKTA_API_TOKEN"]

def okta_headers():
    return {
        "Authorization": f"SSWS {API_TOKEN}",
        "Content-Type": "application/json"
    }

def test_user_provisioning_workflow_triggers_on_group_add():
    """Adding a user to the 'Engineering' group triggers the provisioning workflow."""
    # Create a test user
    user_payload = {
        "profile": {
            "firstName": "Test",
            "lastName": "WorkflowUser",
            "email": "workflow-test@example.com",
            "login": "workflow-test@example.com"
        },
        "credentials": {
            "password": {"value": "TestPass123!"}
        }
    }
    
    create_r = httpx.post(
        f"{OKTA_ORG}/api/v1/users",
        json=user_payload,
        headers=okta_headers()
    )
    assert create_r.status_code in (200, 201)
    user_id = create_r.json()["id"]
    
    try:
        # Find the Engineering group
        groups_r = httpx.get(
            f"{OKTA_ORG}/api/v1/groups",
            params={"q": "Engineering"},
            headers=okta_headers()
        )
        groups = groups_r.json()
        engineering_group = next(
            (g for g in groups if g["profile"]["name"] == "Engineering"),
            None
        )
        assert engineering_group, "Engineering group not found"
        
        # Add user to group — this should trigger the workflow
        add_r = httpx.put(
            f"{OKTA_ORG}/api/v1/groups/{engineering_group['id']}/users/{user_id}",
            headers=okta_headers()
        )
        assert add_r.status_code == 204
        
        # Wait for workflow to execute
        time.sleep(5)
        
        # Verify the workflow's expected side effects
        # (e.g., the user should be assigned to specific apps)
        apps_r = httpx.get(
            f"{OKTA_ORG}/api/v1/users/{user_id}/appLinks",
            headers=okta_headers()
        )
        app_names = [app["label"] for app in apps_r.json()]
        
        # The provisioning workflow should have assigned GitHub and Jira
        assert "GitHub" in app_names, \
            f"GitHub not assigned by provisioning workflow. Apps: {app_names}"
        assert "Jira" in app_names, \
            f"Jira not assigned by provisioning workflow. Apps: {app_names}"
    
    finally:
        # Cleanup: deactivate and delete test user
        httpx.post(
            f"{OKTA_ORG}/api/v1/users/{user_id}/lifecycle/deactivate",
            headers=okta_headers()
        )
        httpx.delete(f"{OKTA_ORG}/api/v1/users/{user_id}", headers=okta_headers())

def test_offboarding_workflow_revokes_access_on_deactivation():
    """Deactivating a user triggers the offboarding workflow."""
    # Create and activate a test user
    user_r = httpx.post(
        f"{OKTA_ORG}/api/v1/users",
        json={
            "profile": {
                "firstName": "Offboard",
                "lastName": "Test",
                "email": "offboard-test@example.com",
                "login": "offboard-test@example.com"
            },
            "credentials": {"password": {"value": "TestPass123!"}}
        },
        params={"activate": "true"},
        headers=okta_headers()
    )
    user_id = user_r.json()["id"]
    
    try:
        # Deactivate — triggers offboarding workflow
        deactivate_r = httpx.post(
            f"{OKTA_ORG}/api/v1/users/{user_id}/lifecycle/deactivate",
            headers=okta_headers()
        )
        assert deactivate_r.status_code == 200
        
        # Wait for workflow
        time.sleep(5)
        
        # Verify user's app assignments are revoked
        apps_r = httpx.get(
            f"{OKTA_ORG}/api/v1/users/{user_id}/appLinks",
            headers=okta_headers()
        )
        # Deactivated user should have no active app assignments
        active_apps = [a for a in apps_r.json() if a.get("status") == "ACTIVE"]
        assert len(active_apps) == 0, \
            f"Offboarding workflow did not revoke apps: {[a['label'] for a in active_apps]}"
    
    finally:
        httpx.delete(f"{OKTA_ORG}/api/v1/users/{user_id}", headers=okta_headers())

# tests/integration/test_okta_event_hooks.py
import pytest
import httpx
import os
import time
import json
from cryptography.hazmat.primitives import hashes, hmac
from cryptography.hazmat.backends import default_backend
import base64

def verify_okta_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify Okta's HMAC-SHA256 webhook signature."""
    h = hmac.HMAC(secret.encode(), hashes.SHA256(), backend=default_backend())
    h.update(payload)
    expected = base64.b64encode(h.finalize()).decode()
    return expected == signature

def test_webhook_signature_verification():
    """Webhook handler correctly verifies Okta's HMAC signature."""
    secret = "test-webhook-secret"
    payload = json.dumps({
        "eventType": "user.session.start",
        "data": {"events": [{"actor": {"id": "user-123"}}]}
    }).encode()
    
    # Compute valid signature
    h = hmac.HMAC(secret.encode(), hashes.SHA256(), backend=default_backend())
    h.update(payload)
    valid_sig = base64.b64encode(h.finalize()).decode()
    
    # Valid signature should pass
    assert verify_okta_webhook_signature(payload, valid_sig, secret)
    
    # Tampered signature should fail
    assert not verify_okta_webhook_signature(payload, "badsignature", secret)

def test_webhook_handler_processes_login_event(test_client):
    """Application webhook handler processes Okta login events correctly."""
    event = {
        "eventType": "user.session.start",
        "published": "2026-05-19T10:00:00.000Z",
        "data": {
            "events": [{
                "eventType": "user.session.start",
                "actor": {
                    "id": "00u1abc2def",
                    "type": "User",
                    "alternateId": "user@example.com",
                    "displayName": "Test User"
                },
                "target": [{
                    "id": "0oa1abc2def",
                    "type": "AppInstance",
                    "displayName": "My App"
                }]
            }]
        }
    }
    
    payload = json.dumps(event).encode()
    
    # Compute signature
    import hmac as hmac_stdlib, hashlib
    sig = base64.b64encode(
        hmac_stdlib.new(b"webhook-secret", payload, hashlib.sha256).digest()
    ).decode()
    
    r = test_client.post(
        "/webhooks/okta",
        data=payload,
        content_type="application/json",
        headers={"x-okta-verification-challenge": sig}
    )
    
    assert r.status_code == 200
    
    # Verify the event was recorded
    events_r = test_client.get("/api/audit/login-events")
    events = events_r.json()["events"]
    
    assert any(
        e["userId"] == "00u1abc2def" and e["email"] == "user@example.com"
        for e in events
    ), "Login event not recorded by webhook handler"

# Test that Okta issuer endpoint is responsive
GET https://dev-12345.oktapreview.com/oauth2/default/.well-known/openid-configuration
Status Should Be 200
Page Should Contain "issuer"
Page Should Contain "jwks_uri"

# Test that your app's login page loads
Go To https://your-app.example.com/login
Page Should Contain "Sign In"