Auth: support RS256 OIDC tokens from bincio-auth

This commit is contained in:
Davide Scaini
2026-06-03 21:24:03 +02:00
parent 3d09097eb3
commit 62bb474908
2 changed files with 38 additions and 8 deletions
+1
View File
@@ -7,4 +7,5 @@ dependencies = [
"uvicorn[standard]>=0.29", "uvicorn[standard]>=0.29",
"pydantic>=2.0", "pydantic>=2.0",
"PyJWT>=2.8", "PyJWT>=2.8",
"cryptography>=42",
] ]
+37 -8
View File
@@ -1,13 +1,15 @@
"""BincioPlanner API — save/load route plans. """BincioPlanner API — save/load route plans.
Auth: validates bincio_session JWT issued by bincio-auth (HS256, shared secret). Auth: validates bincio_session JWT from bincio-auth.
Accepts RS256 OIDC tokens (BINCIO_OIDC_ISSUER) and HS256 session tokens (BINCIO_AUTH_JWT_SECRET).
Storage: JSON files at $PLANS_DIR/{handle}/{plan_id}.json Storage: JSON files at $PLANS_DIR/{handle}/{plan_id}.json
Collections: $PLANS_DIR/{handle}/_collections.json Collections: $PLANS_DIR/{handle}/_collections.json
Shared plans: $PLANS_DIR/_shared/{plan_id}.json (any authed user can read/write/delete) Shared plans: $PLANS_DIR/_shared/{plan_id}.json (any authed user can read/write/delete)
Run: uvicorn server:app --host 127.0.0.1 --port 8002 Run: uvicorn server:app --host 127.0.0.1 --port 8002
Env vars: Env vars:
BINCIO_AUTH_JWT_SECRET HS256 secret shared with bincio-auth (required in production) BINCIO_OIDC_ISSUER OIDC issuer URL — JWKS fetched from {issuer}/.well-known/jwks.json
BINCIO_AUTH_JWT_SECRET HS256 fallback secret
PLANS_DIR root dir for plan JSON files (default: /var/bincio_planner) PLANS_DIR root dir for plan JSON files (default: /var/bincio_planner)
DEV_HANDLE bypass auth in local dev DEV_HANDLE bypass auth in local dev
""" """
@@ -18,23 +20,39 @@ import os
import re import re
import secrets import secrets
import time import time
import urllib.request
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
import jwt as _jwt import jwt as _jwt
from jwt.algorithms import RSAAlgorithm as _RSAAlgorithm
from fastapi import Cookie, Depends, FastAPI, HTTPException, Query from fastapi import Cookie, Depends, FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from pydantic import BaseModel from pydantic import BaseModel
_JWT_SECRET = os.environ.get("BINCIO_AUTH_JWT_SECRET", "") _JWT_SECRET = os.environ.get("BINCIO_AUTH_JWT_SECRET", "")
_PLANS_DIR = Path(os.environ.get("PLANS_DIR", "/var/bincio_planner")) _OIDC_ISSUER = os.environ.get("BINCIO_OIDC_ISSUER", "")
_DEV_HANDLE = os.environ.get("DEV_HANDLE", "") _PLANS_DIR = Path(os.environ.get("PLANS_DIR", "/var/bincio_planner"))
_DEV_HANDLE = os.environ.get("DEV_HANDLE", "")
_SAFE_ID = re.compile(r"^[A-Za-z0-9_-]{1,32}$") _SAFE_ID = re.compile(r"^[A-Za-z0-9_-]{1,32}$")
# Load RS256 public key from JWKS at startup.
_rs256_key = None
if _OIDC_ISSUER:
try:
with urllib.request.urlopen(f"{_OIDC_ISSUER}/.well-known/jwks.json", timeout=5) as _r:
_jwks = json.loads(_r.read())
for _k in _jwks.get("keys", []):
if _k.get("kty") == "RSA":
_rs256_key = _RSAAlgorithm.from_jwk(json.dumps(_k))
break
except Exception as _e:
print(f"Warning: could not load JWKS from {_OIDC_ISSUER}: {_e}")
# ── Auth ─────────────────────────────────────────────────────────────────────── # ── Auth ───────────────────────────────────────────────────────────────────────
@@ -45,6 +63,19 @@ class User:
def _decode_session(token: str) -> Optional[User]: def _decode_session(token: str) -> Optional[User]:
# Try RS256 (OIDC id_token from bincio-auth)
if _rs256_key:
try:
payload = _jwt.decode(token, _rs256_key, algorithms=["RS256"],
options={"verify_aud": False})
handle = payload.get("sub")
if handle and payload.get("activity_access"):
return User(handle=handle,
display_name=payload.get("name") or payload.get("display_name") or handle)
except _jwt.PyJWTError:
pass
# Fall back to HS256 session token
if not _JWT_SECRET: if not _JWT_SECRET:
return None return None
try: try:
@@ -52,9 +83,7 @@ def _decode_session(token: str) -> Optional[User]:
except _jwt.PyJWTError: except _jwt.PyJWTError:
return None return None
handle = payload.get("sub") handle = payload.get("sub")
if not handle: if not handle or not payload.get("activity_access"):
return None
if not payload.get("activity_access"):
return None return None
return User(handle=handle, display_name=payload.get("display_name", handle)) return User(handle=handle, display_name=payload.get("display_name", handle))