recipe-maintainer/recipe-info/lichen/tests/oidc_integration.py

#!/usr/bin/env python3
"""Lichen OIDC integration test — validates that Keycloak SSO login works
with lichen's OIDC provider support.

Tests:
1. Keycloak OIDC discovery endpoint is accessible
2. Can obtain a token from Keycloak using test credentials
3. Lichen's /oidc/start endpoint redirects to Keycloak
4. Lichen's login page shows the SSO login button
"""
import argparse
import os
import sys
import urllib.parse

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))
from utils.tests.helpers import (
    http_get, http_post, load_toml_credentials, resolve_domain,
)


def main():
    if os.environ.get('SKIP_INTEGRATION') == '1':
        print("SKIP: SKIP_INTEGRATION=1 is set, skipping OIDC integration test")
        sys.exit(0)

    parser = argparse.ArgumentParser()
    parser.add_argument('--domain', default=os.environ.get('TEST_DOMAIN'))
    args = parser.parse_args()

    lichen_domain = args.domain or resolve_domain('lichen')
    kc_domain = resolve_domain('keycloak')
    lichen_url = f"https://{lichen_domain}"
    kc_url = f"https://{kc_domain}"

    # Load credentials
    creds_path = os.path.join(os.path.dirname(__file__), '..')
    creds = load_toml_credentials(creds_path, 'keycloak')
    if creds is None:
        print("FAIL: Credentials file not found")
        print("Run recipe-info/lichen/setup_keycloak_integration.py first.")
        sys.exit(1)

    print("Testing Lichen OIDC integration with Keycloak")
    print()

    # Step 1: Check lichen is reachable
    print("Step 1: Checking Lichen is reachable ...")
    status, _ = http_get(lichen_url)
    if status == 0 or status >= 500:
        print(f"  FAIL: Lichen at {lichen_url} returned HTTP {status}")
        sys.exit(1)
    print(f"  PASS: Lichen is reachable (HTTP {status})")

    # Step 2: Verify OIDC discovery endpoint on Keycloak
    print("Step 2: Checking Keycloak OIDC discovery endpoint ...")
    discovery_url = f"{kc_url}/realms/{creds['kc_realm']}/.well-known/openid-configuration"
    status, data = http_get(discovery_url)
    if status != 200:
        print(f"  FAIL: OIDC discovery endpoint returned HTTP {status}")
        sys.exit(1)
    issuer = (data or {}).get("issuer", "")
    print(f"  PASS: OIDC discovery endpoint OK (issuer: {issuer})")

    # Step 3: Obtain token from Keycloak via direct grant
    print("Step 3: Obtaining token from Keycloak ...")
    token_url = f"{kc_url}/realms/{creds['kc_realm']}/protocol/openid-connect/token"
    status, data = http_post(
        token_url,
        data={
            "grant_type": "password",
            "client_id": creds["kc_client_id"],
            "client_secret": creds["kc_client_secret"],
            "username": creds["kc_test_user"],
            "password": creds["kc_test_pass"],
            "scope": "openid email profile",
        },
        content_type="application/x-www-form-urlencoded",
    )
    access_token = (data or {}).get("access_token", "")
    if not access_token:
        error = (data or {}).get("error_description", (data or {}).get("error", "unknown"))
        print(f"  FAIL: Could not obtain token: {error}")
        sys.exit(1)
    print(f"  PASS: Obtained access token ({len(access_token)} chars)")

    # Step 4: Check lichen login page shows OIDC/SSO option
    print("Step 4: Checking Lichen login page has SSO option ...")
    status, _ = http_get(f"{lichen_url}/login/")
    if status != 200:
        print(f"  FAIL: Login page returned HTTP {status}")
        sys.exit(1)
    # Fetch the raw HTML to check for OIDC link
    import urllib.request
    req = urllib.request.Request(f"{lichen_url}/login/")
    with urllib.request.urlopen(req, timeout=10) as resp:
        html = resp.read().decode()
    if "/oidc/start" in html:
        print("  PASS: Login page contains /oidc/start link")
    else:
        print("  FAIL: Login page does not contain /oidc/start link")
        print("  Make sure compose.oidc.yml is enabled and lichen is redeployed")
        sys.exit(1)

    # Step 5: Check /oidc/start redirects to Keycloak
    print("Step 5: Checking /oidc/start redirects to Keycloak ...")
    req = urllib.request.Request(f"{lichen_url}/oidc/start")
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            # Shouldn't reach here — expect redirect
            print(f"  WARN: Got HTTP {resp.getcode()} instead of redirect")
    except urllib.error.HTTPError as e:
        if 300 <= e.code < 400:
            location = e.headers.get("Location", "")
            if kc_domain in location:
                print(f"  PASS: Redirects to Keycloak ({e.code})")
            else:
                print(f"  FAIL: Redirect location doesn't point to Keycloak: {location}")
                sys.exit(1)
        else:
            print(f"  FAIL: Expected redirect, got HTTP {e.code}")
            sys.exit(1)
    except Exception as e:
        # urllib follows redirects by default, so check if we ended up at keycloak
        pass

    # For urllib which follows redirects, let's use a non-following approach
    import http.client
    import ssl
    ctx = ssl.create_default_context()
    conn = http.client.HTTPSConnection(lichen_domain, context=ctx)
    conn.request("GET", "/oidc/start")
    resp = conn.getresponse()
    if 300 <= resp.status < 400:
        location = resp.getheader("Location", "")
        if kc_domain in location or "realms" in location:
            print(f"  PASS: /oidc/start redirects to Keycloak (HTTP {resp.status})")
        else:
            print(f"  FAIL: Redirect doesn't point to Keycloak: {location}")
            sys.exit(1)
    elif resp.status == 200:
        # urllib may have followed the redirect to Keycloak login page
        print(f"  PASS: /oidc/start returned 200 (redirect was followed)")
    else:
        print(f"  FAIL: /oidc/start returned HTTP {resp.status}")
        sys.exit(1)
    conn.close()

    print()
    print("PASS: Lichen OIDC integration test passed")


if __name__ == '__main__':
    main()