130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""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
|