101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
def _normalize_base_url(base_url: str) -> str:
|
|
return base_url.rstrip("/")
|
|
|
|
|
|
def check_oidc(
|
|
base_url: str,
|
|
*,
|
|
require_enabled: bool = False,
|
|
force_health: bool = False,
|
|
timeout: float = 5.0,
|
|
client: httpx.Client | None = None,
|
|
) -> dict[str, Any]:
|
|
base_url = _normalize_base_url(base_url)
|
|
own_client = client is None
|
|
client = client or httpx.Client(timeout=timeout, follow_redirects=False)
|
|
|
|
try:
|
|
config_resp = client.get(f"{base_url}/proxy/auth/auth/oidc/config")
|
|
config_resp.raise_for_status()
|
|
config = config_resp.json()
|
|
enabled = bool(config.get("enabled"))
|
|
|
|
if require_enabled and not enabled:
|
|
raise RuntimeError("OIDC is disabled but --require-enabled was specified")
|
|
|
|
summary: dict[str, Any] = {
|
|
"base_url": base_url,
|
|
"config": config,
|
|
"health": None,
|
|
}
|
|
|
|
if enabled or force_health:
|
|
health_resp = client.get(f"{base_url}/proxy/auth/auth/oidc/health")
|
|
health_resp.raise_for_status()
|
|
summary["health"] = health_resp.json()
|
|
|
|
return summary
|
|
finally:
|
|
if own_client:
|
|
client.close()
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Smoke-check OIDC config and provider health via the gateway")
|
|
parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway base URL")
|
|
parser.add_argument(
|
|
"--require-enabled",
|
|
action="store_true",
|
|
help="Fail if OIDC is disabled",
|
|
)
|
|
parser.add_argument(
|
|
"--force-health",
|
|
action="store_true",
|
|
help="Call the OIDC health endpoint even if OIDC is disabled",
|
|
)
|
|
parser.add_argument("--timeout", type=float, default=5.0, help="HTTP timeout in seconds")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = _build_parser()
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
summary = check_oidc(
|
|
args.base_url,
|
|
require_enabled=args.require_enabled,
|
|
force_health=args.force_health,
|
|
timeout=args.timeout,
|
|
)
|
|
except (httpx.HTTPError, RuntimeError) as exc:
|
|
print(f"OIDC smoke check failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
config = summary["config"]
|
|
health = summary["health"]
|
|
|
|
print("OIDC config:")
|
|
print(json.dumps(config, ensure_ascii=False, indent=2))
|
|
if health is not None:
|
|
print("OIDC health:")
|
|
print(json.dumps(health, ensure_ascii=False, indent=2))
|
|
else:
|
|
print("OIDC health skipped because OIDC is disabled")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|