LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Stricter security. Moved to /auth/oidc/

Browse Source
This commit is contained in:
Leo Vasanko
2026-02-15 02:57:23 +00:00
parent b5a5f2707a
commit 701b0810bd
5 changed files with 61 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oidc.md
+11 -11
View File
@@ -48,9 +48,9 @@ oid_auth_codes: dict[str, OIDAuthCode] = {}
- `GET /.well-known/openid-configuration` — Discovery document
- `GET /.well-known/jwks.json` — Public keys for token verification
### OIDC routes (`/auth/oid/`)
- `POST /auth/oid/token` — Token endpoint (code exchange)
- `GET /auth/oid/userinfo` — UserInfo endpoint (bearer token)
### OIDC routes (`/auth/oidc/`)
- `POST /auth/oidc/token` — Token endpoint (code exchange)
- `GET /auth/oidc/userinfo` — UserInfo endpoint (bearer token)
### Authorization (via existing restricted app)
- `GET /auth/restricted/?client_id=...&redirect_uri=...&response_type=code&scope=openid...`
@@ -58,8 +58,8 @@ oid_auth_codes: dict[str, OIDAuthCode] = {}
The restricted app detects OIDC params from URL and handles authentication via WebSocket.
## JWT & Signing
- RSA keypair generated on first boot (stored in data directory)
- ID tokens signed with RS256
- Ed25519 keypair generated on first boot (stored in data directory)
- ID tokens signed with EdDSA
- `kid` in JWKS for key rotation support
- Access tokens are signed JWTs (not opaque)
@@ -87,19 +87,19 @@ The `/auth/restricted/` page handles OIDC authorization alongside normal iframe
1. Client redirects to `/auth/restricted/?client_id=...&redirect_uri=...&response_type=code&scope=openid&state=...`
2. Frontend detects OIDC params from `window.location.search`
3. User authenticates via passkey
4. Frontend passes raw query string to `/auth/ws/authenticate?{query_string}`
3. Frontend passes raw query string to `/auth/ws/authenticate?{query_string}`
4. User authenticates via passkey
5. WebSocket validates client/redirect_uri, authenticates user, creates auth code
6. WebSocket returns `{"redirect_url": "redirect_uri?code=...&state=..."}`
7. Frontend redirects to the URL
8. Client exchanges code at `/auth/oid/token` → receives `id_token` + `access_token`
9. Optionally calls `/auth/oid/userinfo` with bearer token
8. Client exchanges code at `/auth/oidc/token` → receives `id_token` + `access_token`
9. Optionally calls `/auth/oidc/userinfo` with bearer token
**Key design points:**
- No session created during OIDC auth (stateless for the OIDC client)
- Raw query string preserved throughout (no parsing/reconstruction of redirect_uri)
- Redirect URI validated against client's registered URIs via exact string match
- PKCE supported (S256 and plain methods)
- PKCE required (S256 only)
## WebSocket OIDC Mode
@@ -124,7 +124,7 @@ When OIDC params absent:
- `paskia/db/structs.py` — Added `OIDClient`, `OIDAuthCode` models; User fields `email`, `preferred_username`
- `paskia/db/operations.py` — CRUD for OIDC entities
- `paskia/fastapi/oid.py` — Token and userinfo endpoints
- `paskia/util/oidjwt.py` — RSA key management, JWT creation, JWKS
- `paskia/util/oidjwt.py` — Ed25519 key management, JWT creation, JWKS
### Modified
- `paskia/fastapi/mainapp.py` — Mount OIDC app, well-known endpoints
paskia/fastapi/mainapp.py
+5 -4
View File
@@ -88,7 +88,7 @@ app.middleware("http")(auth_host.redirect_middleware)
app.mount("/auth/api/admin/", admin.app)
app.mount("/auth/api/", api.app)
app.mount("/auth/ws/", ws.app)
app.mount("/auth/oid/", oid.app)
app.mount("/auth/oidc/", oid.app)
# OIDC Well-Known endpoints (must be at site root)
@@ -103,17 +103,18 @@ async def openid_configuration(request: Request):
return {
"issuer": issuer,
"authorization_endpoint": f"{issuer}/auth/restricted/",
"token_endpoint": f"{issuer}/auth/oid/token",
"userinfo_endpoint": f"{issuer}/auth/oid/userinfo",
"token_endpoint": f"{issuer}/auth/oidc/token",
"userinfo_endpoint": f"{issuer}/auth/oidc/userinfo",
"jwks_uri": f"{issuer}/.well-known/jwks.json",
"response_types_supported": ["code"],
"subject_types_supported": ["public"],
"id_token_signing_alg_values_supported": ["RS256"],
"id_token_signing_alg_values_supported": ["EdDSA"],
"scopes_supported": ["openid", "profile", "email"],
"token_endpoint_auth_methods_supported": [
"client_secret_post",
"client_secret_basic",
],
"code_challenge_methods_supported": ["S256"],
"claims_supported": [
"sub",
"name",
paskia/fastapi/oid.py
+15 -12
View File
@@ -33,16 +33,11 @@ def _get_issuer(request: Request) -> str:
return f"{scheme}://{host}"
def _verify_pkce(code_verifier: str, code_challenge: str, method: str) -> bool:
"""Verify PKCE code_verifier against stored code_challenge."""
if method == "plain":
return code_verifier == code_challenge
elif method == "S256":
# SHA256 hash, base64url encode
digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
computed = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
return computed == code_challenge
return False
def _verify_pkce(code_verifier: str, code_challenge: str) -> bool:
"""Verify PKCE code_verifier against stored code_challenge (S256 only)."""
digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
computed = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
return computed == code_challenge
def _parse_client_credentials(
@@ -147,8 +142,16 @@ async def token(
},
status_code=400,
)
method = auth_code.code_challenge_method or "plain"
if not _verify_pkce(code_verifier, auth_code.code_challenge, method):
method = auth_code.code_challenge_method or "S256"
if method != "S256":
return JSONResponse(
{
"error": "invalid_grant",
"error_description": "Only S256 code_challenge_method is supported",
},
status_code=400,
)
if not _verify_pkce(code_verifier, auth_code.code_challenge):
return JSONResponse(
{
"error": "invalid_grant",
paskia/fastapi/ws.py
+15
View File
@@ -129,6 +129,21 @@ async def websocket_authenticate(
await ws.send_json({"status": 400, "detail": "Scope must include openid"})
return
# PKCE is required with S256
if not code_challenge:
await ws.send_json(
{"status": 400, "detail": "PKCE code_challenge is required"}
)
return
if code_challenge_method and code_challenge_method != "S256":
await ws.send_json(
{
"status": 400,
"detail": "Only S256 code_challenge_method is supported",
}
)
return
# If there's an existing session, restrict to that user's credentials (reauth)
session_user_uuid = None
if auth:
paskia/util/oidjwt.py
+15 -22
View File
@@ -11,7 +11,7 @@ from uuid import UUID
import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
_logger = logging.getLogger(__name__)
@@ -25,7 +25,7 @@ _KEY_FILE = Path("oidc_key.pem")
def _load_or_generate_key() -> None:
"""Load existing RSA key or generate a new one."""
"""Load existing Ed25519 key or generate a new one."""
global _private_key, _public_key, _kid
if _KEY_FILE.exists():
@@ -34,10 +34,7 @@ def _load_or_generate_key() -> None:
_private_key = serialization.load_pem_private_key(pem_data, password=None)
else:
_logger.info("Generating new OIDC signing key")
_private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
)
_private_key = Ed25519PrivateKey.generate()
pem_data = _private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
@@ -61,28 +58,24 @@ def _ensure_key() -> None:
_load_or_generate_key()
def _b64url_uint(value: int) -> str:
"""Encode an integer as base64url without padding (for JWK)."""
# Calculate minimum bytes needed
byte_length = (value.bit_length() + 7) // 8
value_bytes = value.to_bytes(byte_length, byteorder="big")
return urlsafe_b64encode(value_bytes).rstrip(b"=").decode("ascii")
def get_jwks() -> dict:
"""Get JWKS (JSON Web Key Set) for public key verification."""
_ensure_key()
assert _public_key is not None
pub_numbers = _public_key.public_numbers()
# Ed25519 public key is 32 bytes raw
pub_bytes = _public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw,
)
return {
"keys": [
{
"kty": "RSA",
"kty": "OKP",
"crv": "Ed25519",
"use": "sig",
"alg": "RS256",
"alg": "EdDSA",
"kid": _kid,
"n": _b64url_uint(pub_numbers.n),
"e": _b64url_uint(pub_numbers.e),
"x": urlsafe_b64encode(pub_bytes).rstrip(b"=").decode("ascii"),
}
]
}
@@ -135,7 +128,7 @@ def create_id_token(
if permissions:
payload["permissions"] = permissions
return jwt.encode(payload, _private_key, algorithm="RS256", headers={"kid": _kid})
return jwt.encode(payload, _private_key, algorithm="EdDSA", headers={"kid": _kid})
def create_access_token(
@@ -167,7 +160,7 @@ def create_access_token(
"iat": int(now.timestamp()),
"exp": int((now + timedelta(seconds=expires_in)).timestamp()),
}
return jwt.encode(payload, _private_key, algorithm="RS256", headers={"kid": _kid})
return jwt.encode(payload, _private_key, algorithm="EdDSA", headers={"kid": _kid})
def decode_access_token(token: str, issuer: str) -> dict | None:
@@ -185,7 +178,7 @@ def decode_access_token(token: str, issuer: str) -> dict | None:
return jwt.decode(
token,
_public_key,
algorithms=["RS256"],
algorithms=["EdDSA"],
issuer=issuer,
options={"verify_aud": False}, # We'll verify audience separately if needed
)