bvandeusen
af60ca446d
The fabledscryer->steward rename had only ever reached host_agent. The other five bundled plugins (http, snmp, traefik, unifi, docker) still imported `from fabledscryer.*` (package no longer exists) and read FABLEDSCRYER_* env vars — so every one of them was broken at import since the original rebrand. CI stayed green only because none are enabled by default and migrations don't import plugin modules. Now that they version in-tree, complete the rename: - fabledscryer.* -> steward.* imports across all five plugins - FABLEDSCRYER_* -> STEWARD_* in plugin migration env.py files - author/repository/homepage + user-facing 'Fabled Scryer' strings -> Steward - snmp/scheduler.py: also drop dead `now`/datetime; record_metric from steward Adds tests/test_no_legacy_names.py — fails if 'scryer'/'roundtable' ever reappear in shipped code (the drift bit twice; this stops a third time). Also clears pre-existing ruff lint debt (unused imports, semicolon statements, mid-file import) surfaced by the new lint lane. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""steward/auth/oidc.py
|
|
|
|
OIDC Authorization Code flow helpers using httpx.
|
|
No extra dependencies — works with any OIDC provider (Authentik, Keycloak, etc.).
|
|
|
|
JWT signature verification is intentionally skipped for homelab use;
|
|
we trust the HTTPS connection to the discovery-documented token/userinfo endpoints.
|
|
"""
|
|
from __future__ import annotations
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Module-level cache: {discovery_url: (fetched_at, doc)}
|
|
_discovery_cache: dict[str, tuple[float, dict]] = {}
|
|
_DISCOVERY_TTL = 300 # seconds
|
|
|
|
|
|
async def get_discovery(discovery_url: str) -> dict:
|
|
"""Fetch and cache the OIDC provider discovery document."""
|
|
now = time.monotonic()
|
|
if discovery_url in _discovery_cache:
|
|
fetched_at, doc = _discovery_cache[discovery_url]
|
|
if now - fetched_at < _DISCOVERY_TTL:
|
|
return doc
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(discovery_url)
|
|
resp.raise_for_status()
|
|
doc = resp.json()
|
|
_discovery_cache[discovery_url] = (now, doc)
|
|
return doc
|
|
|
|
|
|
def build_authorize_url(doc: dict, client_id: str, redirect_uri: str,
|
|
scopes: str, state: str, nonce: str) -> str:
|
|
"""Build the IdP authorization redirect URL."""
|
|
import urllib.parse
|
|
params = {
|
|
"response_type": "code",
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": scopes,
|
|
"state": state,
|
|
"nonce": nonce,
|
|
}
|
|
return doc["authorization_endpoint"] + "?" + urllib.parse.urlencode(params)
|
|
|
|
|
|
async def exchange_code(doc: dict, client_id: str, client_secret: str,
|
|
code: str, redirect_uri: str) -> dict:
|
|
"""Exchange authorization code for tokens. Returns token response dict."""
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.post(
|
|
doc["token_endpoint"],
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def get_userinfo(doc: dict, access_token: str) -> dict:
|
|
"""Fetch user info from the IdP's userinfo endpoint."""
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
doc["userinfo_endpoint"],
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def decode_id_token_payload(id_token: str) -> dict:
|
|
"""Decode ID token payload without verifying signature (homelab only)."""
|
|
try:
|
|
parts = id_token.split(".")
|
|
if len(parts) < 2:
|
|
return {}
|
|
payload_b64 = parts[1]
|
|
# Add padding
|
|
payload_b64 += "=" * (4 - len(payload_b64) % 4)
|
|
return json.loads(base64.urlsafe_b64decode(payload_b64))
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def map_role(groups: list[str], admin_group: str, operator_group: str) -> str:
|
|
"""Map IdP groups to a local role string."""
|
|
if admin_group and admin_group in groups:
|
|
return "admin"
|
|
if operator_group and operator_group in groups:
|
|
return "operator"
|
|
return "viewer"
|
|
|
|
|
|
def generate_state() -> str:
|
|
return base64.urlsafe_b64encode(os.urandom(24)).decode().rstrip("=")
|