build v_1.0.1

ipsec_migrator/keystone_auth.py

@@ -0,0 +1,129 @@
+"""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