auth: add RS256 validation via JWKS (Phase 3)
CI / Python tests (push) Waiting to run
CI / Frontend build (push) Waiting to run

When --oidc-issuer is set, validate tokens as RS256 id_tokens fetched
against bincio-auth's JWKS endpoint (cached for 1h). Falls back to
HS256 if the RS256 check fails, so existing sessions keep working
during the transition. DB session lookup is the final fallback.

New --oidc-issuer flag reads BINCIO_OIDC_ISSUER env var.
This commit is contained in:
Davide Scaini
2026-06-03 15:43:44 +02:00
parent e24d290127
commit 3394be4ee9
2 changed files with 83 additions and 9 deletions
+74 -6
View File
@@ -38,12 +38,20 @@ dem_url: str = "https://api.open-elevation.com"
sync_secret: str = ""
jwt_secret: str = "" # when set, validates JWTs from bincio-auth instead of DB session lookup
auth_api: str = "" # when set, proxies user-state admin ops to bincio-auth (e.g. http://127.0.0.1:4040)
oidc_issuer: str = "" # when set, validates RS256 id_tokens via bincio-auth JWKS
_db = None
_strava_sync_running = False
_strava_sync_lock = threading.Lock()
_garmin_sync_running = False
_garmin_sync_lock = threading.Lock()
# ── JWKS cache ────────────────────────────────────────────────────────────────
_jwks_public_key: object = None
_jwks_fetched_at: float = 0.0
_jwks_lock = threading.Lock()
_JWKS_TTL = 3600
# ── Constants ─────────────────────────────────────────────────────────────────
_VALID_HANDLE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,29}$')
@@ -116,8 +124,59 @@ def _check_rate_limit(
# ── Auth dependency functions ─────────────────────────────────────────────────
def _decode_jwt(token: str) -> User | None:
"""Decode a bincio-auth JWT and return a User. Returns None on any failure."""
def _get_jwks_public_key() -> object:
"""Fetch and cache the RSA public key from bincio-auth's JWKS endpoint."""
global _jwks_public_key, _jwks_fetched_at
now = time.time()
with _jwks_lock:
if _jwks_public_key is not None and now - _jwks_fetched_at < _JWKS_TTL:
return _jwks_public_key
import base64
import urllib.request
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers
from cryptography.hazmat.backends import default_backend
url = f"{oidc_issuer.rstrip('/')}/.well-known/jwks.json"
with urllib.request.urlopen(url, timeout=5) as r:
jwks = json.loads(r.read())
k = jwks["keys"][0]
def _b64i(s: str) -> int:
s += "=" * (-len(s) % 4)
return int.from_bytes(base64.urlsafe_b64decode(s), "big")
pub = RSAPublicNumbers(_b64i(k["e"]), _b64i(k["n"])).public_key(default_backend())
_jwks_public_key = pub
_jwks_fetched_at = now
return pub
def _decode_rs256(token: str) -> User | None:
"""Decode an RS256 id_token from bincio-auth. Returns None on any failure."""
try:
pub = _get_jwks_public_key()
payload = _jwt.decode(
token, pub, algorithms=["RS256"],
options={"verify_aud": False},
issuer=oidc_issuer,
)
except Exception:
return None
handle = payload.get("sub")
if not handle:
return None
return User(
handle=handle,
display_name=payload.get("name") or payload.get("display_name", ""),
is_admin=bool(payload.get("is_admin", False)),
wiki_access=bool(payload.get("wiki_access", True)),
activity_access=bool(payload.get("activity_access", False)),
suspended=False,
created_at=0,
)
def _decode_hs256(token: str) -> User | None:
"""Decode a bincio-auth HS256 JWT and return a User. Returns None on any failure."""
try:
payload = _jwt.decode(token, jwt_secret, algorithms=["HS256"])
except _jwt.PyJWTError:
@@ -136,12 +195,21 @@ def _decode_jwt(token: str) -> User | None:
)
def _decode_token(token: str) -> User | None:
"""Try RS256 first (if oidc_issuer set), then HS256, then DB session."""
if oidc_issuer:
user = _decode_rs256(token)
if user:
return user
if jwt_secret:
return _decode_hs256(token)
return get_session(_get_db(), token)
def _current_user(bincio_session: str | None = Cookie(default=None)) -> User | None:
if not bincio_session:
return None
if jwt_secret:
return _decode_jwt(bincio_session)
return get_session(_get_db(), bincio_session)
return _decode_token(bincio_session)
def _require_user(bincio_session: str | None = Cookie(default=None)) -> User:
@@ -170,7 +238,7 @@ def _require_auth(
token = auth[7:]
if not token:
raise HTTPException(401, "Not authenticated")
user = _decode_jwt(token) if jwt_secret else get_session(_get_db(), token)
user = _decode_token(token)
if not user:
raise HTTPException(401, "Invalid or expired session")
return user