VPNaaS-Migrator/ipsec_migrator/keystone_auth.py

"""Keystone v3 password authentication via `requests`, replacing the
`openstack token issue` subprocess call.

Mirrors the auth flow the `openstack` CLI itself performs when reading
`OS_*` environment variables: POST a project-scoped password auth request
to `{OS_AUTH_URL}/auth/tokens`, read the token back from the
`X-Subject-Token` response header, and keep the returned service catalog
around so the Neutron endpoint can be resolved from it (same as the CLI
does) instead of being hardcoded.

No `clouds.yaml` support -- the only auth pattern actually in use here
(`test_openrc.sh`) is password auth via `OS_*` env vars.
"""
from __future__ import annotations

import os
import sys

import requests

_REQUIRED = ("OS_AUTH_URL", "OS_USERNAME", "OS_PASSWORD")


def _env(name: str) -> str | None:
    value = os.environ.get(name)
    return value if value else None


def _user_domain() -> dict:
    if domain_id := _env("OS_USER_DOMAIN_ID"):
        return {"id": domain_id}
    return {"name": _env("OS_USER_DOMAIN_NAME") or "Default"}


def _project_scope() -> dict:
    if project_id := _env("OS_PROJECT_ID"):
        return {"id": project_id}

    project_name = _env("OS_PROJECT_NAME")
    if not project_name:
        print(
            "Error: neither OS_PROJECT_ID nor OS_PROJECT_NAME is set.",
            file=sys.stderr,
        )
        sys.exit(1)

    if project_domain_id := _env("OS_PROJECT_DOMAIN_ID"):
        domain = {"id": project_domain_id}
    else:
        domain = {"name": _env("OS_PROJECT_DOMAIN_NAME") or "Default"}

    return {"name": project_name, "domain": domain}


def authenticate() -> tuple[str, list[dict]]:
    """Returns (token, service_catalog)."""
    missing = [name for name in _REQUIRED if not _env(name)]
    if missing:
        print(f"Error: required environment variable(s) not set: {', '.join(missing)}", file=sys.stderr)
        sys.exit(1)

    auth_url = _env("OS_AUTH_URL").rstrip("/")
    body = {
        "auth": {
            "identity": {
                "methods": ["password"],
                "password": {
                    "user": {
                        "name": _env("OS_USERNAME"),
                        "domain": _user_domain(),
                        "password": _env("OS_PASSWORD"),
                    }
                },
            },
            "scope": {"project": _project_scope()},
        }
    }

    try:
        response = requests.post(f"{auth_url}/auth/tokens", json=body, timeout=30.0)
    except requests.exceptions.RequestException as exc:
        print(f"Error: Keystone authentication request failed: {exc}", file=sys.stderr)
        sys.exit(1)

    if not 200 <= response.status_code < 300:
        print(f"Error: Keystone authentication failed (HTTP {response.status_code})", file=sys.stderr)
        print(f"Response: {response.text}", file=sys.stderr)
        sys.exit(1)

    token = response.headers.get("X-Subject-Token")
    if not token:
        print("Error: Keystone response did not include X-Subject-Token.", file=sys.stderr)
        sys.exit(1)

    catalog = (response.json().get("token") or {}).get("catalog") or []
    return token, catalog


def resolve_endpoint(catalog: list[dict], service_type: str = "network") -> str:
    interface = _env("OS_INTERFACE") or "public"
    region = _env("OS_REGION_NAME")

    service = next((entry for entry in catalog if entry.get("type") == service_type), None)
    if service is None:
        print(f"Error: service type '{service_type}' not found in Keystone catalog.", file=sys.stderr)
        sys.exit(1)

    candidates = [ep for ep in service.get("endpoints", []) if ep.get("interface") == interface]
    if region:
        region_matches = [ep for ep in candidates if ep.get("region") == region]
        if region_matches:
            candidates = region_matches

    if not candidates:
        print(
            f"Error: no '{interface}' endpoint for service type '{service_type}' in Keystone catalog"
            + (f" (region {region})" if region else ""),
            file=sys.stderr,
        )
        sys.exit(1)

    url = candidates[0]["url"].rstrip("/")
    # Neutron's catalog entry is registered with the /v2.0 suffix already on
    # some clouds and without it on others; neutron_client.py always adds
    # /v2.0/... itself, so strip a pre-existing suffix here to avoid
    # double-versioned URLs (.../v2.0/v2.0/routers).
    if service_type == "network" and (url.endswith("/v2.0") or url.endswith("/v2")):
        url = url.rsplit("/", 1)[0]
    return url