280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""SAML 2.0 Service Provider — wrapper around python3-saml (OneLogin toolkit).
|
|
|
|
v6.7.0. Adapts FastAPI's ``Request`` to the toolkit's flat ``request_data``
|
|
dict and builds the SP settings from the ``sso_config`` row.
|
|
|
|
What the toolkit validates in strict mode (all covered by tests):
|
|
XML schema, signature of the assertion and/or the message against the IdP
|
|
certificate, ``Conditions`` timestamps, ``Audience``, ``Destination``,
|
|
``Issuer``, ``Status``, "exactly one assertion", and ``InResponseTo``
|
|
against the AuthnRequest id we pass to ``process_response()`` — combined
|
|
with the single-use ``sso_requests`` store that makes replay impossible.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
|
|
from fastapi import Request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BINDING_HTTP_REDIRECT = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
|
|
BINDING_HTTP_POST = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
|
|
NAMEID_FORMAT_EMAIL = "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
|
|
|
|
#: Default attribute mapping (design doc §3.2). ``nameid`` = the assertion's
|
|
#: NameID; every other value is matched against attribute Name / FriendlyName
|
|
#: / URI local part (so ``email`` finds both ``email`` and
|
|
#: ``http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress``).
|
|
DEFAULT_SAML_MAPPING: dict[str, str] = {
|
|
"login": "nameid",
|
|
"email": "nameid",
|
|
"full_name": "displayName",
|
|
"avatar_url": "avatar",
|
|
"groups": "groups",
|
|
}
|
|
|
|
|
|
class SAMLError(Exception):
|
|
"""SAML processing failure — ``message`` is user-facing, ``reason`` is logged."""
|
|
|
|
|
|
@dataclass
|
|
class SAMLIdentity:
|
|
"""What a validated assertion tells us about the user."""
|
|
|
|
name_id: str
|
|
name_id_format: str = ""
|
|
session_index: str = ""
|
|
attributes: dict[str, list[str]] = field(default_factory=dict)
|
|
friendly_attributes: dict[str, list[str]] = field(default_factory=dict)
|
|
|
|
def resolve(self, source: str) -> str:
|
|
"""Resolve one mapped source (``nameid`` or an attribute name) → first value."""
|
|
if not source or source == "nameid":
|
|
return self.name_id or ""
|
|
if source in self.attributes and self.attributes[source]:
|
|
return (self.attributes[source][0] or "").strip()
|
|
# FriendlyName match (case-insensitive)
|
|
lower = {k.lower(): v for k, v in self.friendly_attributes.items()}
|
|
if source.lower() in lower and lower[source.lower()]:
|
|
return (lower[source.lower()][0] or "").strip()
|
|
# URI local part match: ".../claims/emailaddress" ~ "emailaddress", and
|
|
# a mapping of "email" must still find ".../claims/emailaddress".
|
|
want = source.lower().lstrip("./")
|
|
for name, values in self.attributes.items():
|
|
if not values:
|
|
continue
|
|
local = name.rsplit("/", 1)[-1].rsplit("}", 1)[-1].lower()
|
|
if local == want or local.endswith(want) or want.endswith(local):
|
|
return (values[0] or "").strip()
|
|
return ""
|
|
|
|
|
|
def external_base_url(request: Request) -> str:
|
|
"""Scheme://host the user actually used (proxy-aware, like OAuth redirects)."""
|
|
proto = request.headers.get("x-forwarded-proto", "")
|
|
scheme = proto.split(",")[0].strip() or request.url.scheme or "http"
|
|
fwd_host = request.headers.get("x-forwarded-host", "")
|
|
host = fwd_host.split(",")[0].strip() or request.headers.get("host", "localhost:8080")
|
|
return f"{scheme}://{host}"
|
|
|
|
|
|
def saml_endpoints(request: Request) -> dict[str, str]:
|
|
"""SP entity id + ACS/SLO/metadata URLs derived from the incoming request."""
|
|
base = external_base_url(request)
|
|
return {
|
|
"entity_id": f"{base}/auth/saml/metadata",
|
|
"acs": f"{base}/auth/saml/callback",
|
|
"slo": f"{base}/auth/saml/logout",
|
|
"metadata": f"{base}/auth/saml/metadata",
|
|
}
|
|
|
|
|
|
def build_settings(cfg: dict, endpoints: dict[str, str]) -> dict:
|
|
"""python3-saml settings dict built from a ``sso_config`` row."""
|
|
sign_requests = bool(cfg.get("sign_requests"))
|
|
sp: dict = {
|
|
"entityId": endpoints["entity_id"],
|
|
"assertionConsumerService": {
|
|
"url": endpoints["acs"],
|
|
"binding": BINDING_HTTP_POST,
|
|
},
|
|
"singleLogoutService": {
|
|
"url": endpoints["slo"],
|
|
"binding": BINDING_HTTP_REDIRECT,
|
|
},
|
|
"NameIDFormat": NAMEID_FORMAT_EMAIL,
|
|
}
|
|
if cfg.get("sp_private_key") and cfg.get("sp_certificate"):
|
|
sp["privateKey"] = cfg["sp_private_key"]
|
|
sp["x509cert"] = cfg["sp_certificate"]
|
|
|
|
idp: dict = {
|
|
"entityId": cfg.get("entity_id") or "",
|
|
"singleSignOnService": {
|
|
"url": cfg.get("sso_url") or "",
|
|
"binding": BINDING_HTTP_REDIRECT,
|
|
},
|
|
"x509cert": cfg.get("x509_certificate") or "",
|
|
}
|
|
if cfg.get("slo_url"):
|
|
idp["singleLogoutService"] = {"url": cfg["slo_url"], "binding": BINDING_HTTP_REDIRECT}
|
|
|
|
return {
|
|
"strict": True,
|
|
"debug": False,
|
|
"sp": sp,
|
|
"idp": idp,
|
|
"security": {
|
|
"authnRequestsSigned": sign_requests,
|
|
"logoutRequestSigned": sign_requests,
|
|
"logoutResponseSigned": False,
|
|
"wantMessagesSigned": False,
|
|
"wantAssertionsSigned": True,
|
|
"wantNameIdEncrypted": False,
|
|
"wantAssertionsEncrypted": False,
|
|
"wantXmlValidation": True,
|
|
"signatureAlgorithm": "http://www.w3.org/2001/04/xmldsig-more#rsa-sha256",
|
|
"digestAlgorithm": "http://www.w3.org/2001/04/xmlenc#sha256",
|
|
"rejectDeprecatedAlgorithm": True,
|
|
# FlowDeck is self-hosted: LAN/homelab deploys commonly reach the
|
|
# SP through single-label hosts (http://flowdeck/, docker service
|
|
# names). python3-saml rejects those URLs unless this is on.
|
|
"allowSingleLabelDomains": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _request_data(request: Request, script_name: str, post_data: dict | None = None) -> dict:
|
|
"""Flat request dict expected by ``OneLogin_Saml2_Auth``."""
|
|
https = "on" if external_base_url(request).startswith("https") else "off"
|
|
return {
|
|
"https": https,
|
|
"http_host": request.headers.get("host", "localhost:8080"),
|
|
"script_name": script_name,
|
|
"request_uri": request.url.path,
|
|
"query_string": str(request.url.query or ""),
|
|
"get_data": dict(request.query_params),
|
|
"post_data": post_data or {},
|
|
}
|
|
|
|
|
|
def _auth(request: Request, cfg: dict, script_name: str, post_data: dict | None = None):
|
|
from onelogin.saml2.auth import OneLogin_Saml2_Auth
|
|
|
|
settings = build_settings(cfg, saml_endpoints(request))
|
|
try:
|
|
return OneLogin_Saml2_Auth(
|
|
_request_data(request, script_name, post_data=post_data), old_settings=settings
|
|
)
|
|
except Exception as err: # malformed IdP/SP config (bad cert, missing URL…)
|
|
raise SAMLError(f"Invalid SAML configuration: {err}") from err
|
|
|
|
|
|
def create_login(request: Request, cfg: dict, relay_state: str) -> tuple[str, str]:
|
|
"""Build the AuthnRequest. Returns ``(redirect_url, authn_request_id)``."""
|
|
auth = _auth(request, cfg, "/auth/saml/login")
|
|
try:
|
|
url = auth.login(return_to=relay_state)
|
|
except Exception as err:
|
|
raise SAMLError(f"Could not build the SAML AuthnRequest: {err}") from err
|
|
request_id = auth.get_last_request_id() or ""
|
|
if not request_id:
|
|
raise SAMLError("AuthnRequest was built without an id")
|
|
return url, request_id
|
|
|
|
|
|
def process_response(request: Request, cfg: dict, post_data: dict, request_id: str) -> SAMLIdentity:
|
|
"""Validate the IdP's SAMLResponse and extract the identity.
|
|
|
|
``request_id`` is the id of the AuthnRequest we issued (from the
|
|
single-use ``sso_requests`` row): the toolkit rejects any response whose
|
|
``InResponseTo`` does not match it.
|
|
"""
|
|
auth = _auth(request, cfg, "/auth/saml/callback", post_data=post_data)
|
|
try:
|
|
auth.process_response(request_id=request_id or None)
|
|
except Exception as err:
|
|
raise SAMLError(f"SAML response could not be processed: {err}") from err
|
|
|
|
errors = auth.get_errors()
|
|
if errors:
|
|
raise SAMLError(auth.get_last_error_reason() or f"SAML errors: {', '.join(errors)}")
|
|
if not auth.is_authenticated():
|
|
raise SAMLError("SAML response did not authenticate the user")
|
|
|
|
name_id = auth.get_nameid() or ""
|
|
if not name_id:
|
|
raise SAMLError("SAML assertion carries no NameID")
|
|
return SAMLIdentity(
|
|
name_id=name_id,
|
|
name_id_format=auth.get_nameid_format() or "",
|
|
session_index=auth.get_session_index() or "",
|
|
attributes=auth.get_attributes() or {},
|
|
friendly_attributes=auth.get_friendlyname_attributes() or {},
|
|
)
|
|
|
|
|
|
def metadata_xml(request: Request, cfg: dict) -> str:
|
|
"""SP metadata XML (for the IdP configuration screen)."""
|
|
from onelogin.saml2.settings import OneLogin_Saml2_Settings
|
|
|
|
settings = OneLogin_Saml2_Settings(
|
|
build_settings(cfg, saml_endpoints(request)), custom_base_path=None
|
|
)
|
|
try:
|
|
xml = settings.get_sp_metadata()
|
|
except Exception as err:
|
|
raise SAMLError(f"Could not build the SP metadata: {err}") from err
|
|
if isinstance(xml, bytes):
|
|
xml = xml.decode("utf-8")
|
|
return xml
|
|
|
|
|
|
def build_logout_url(request: Request, cfg: dict, return_to: str, name_id: str, session_index: str) -> str:
|
|
"""SP-initiated Single Logout (HTTP-Redirect LogoutRequest to the IdP)."""
|
|
auth = _auth(request, cfg, "/auth/saml/logout")
|
|
if not cfg.get("slo_url"):
|
|
raise SAMLError("The IdP has no Single Logout URL configured")
|
|
try:
|
|
return auth.logout(
|
|
return_to=return_to,
|
|
name_id=name_id or None,
|
|
session_index=session_index or None,
|
|
)
|
|
except Exception as err:
|
|
raise SAMLError(f"Could not build the SAML LogoutRequest: {err}") from err
|
|
|
|
|
|
def process_slo_form(request: Request, cfg: dict, form: dict, query: dict) -> tuple[str | None, list[str]]:
|
|
"""Process a LogoutRequest / LogoutResponse received from the IdP.
|
|
|
|
``form`` holds the POSTed fields, ``query`` the GET parameters (the
|
|
HTTP-Redirect binding delivers LogoutRequest/LogoutResponse there).
|
|
Returns ``(redirect_url, errors)``.
|
|
"""
|
|
from onelogin.saml2.auth import OneLogin_Saml2_Auth
|
|
|
|
settings = build_settings(cfg, saml_endpoints(request))
|
|
https = "on" if external_base_url(request).startswith("https") else "off"
|
|
post_data = {k: v for k, v in form.items() if k in ("SAMLRequest", "SAMLResponse", "RelayState")}
|
|
if not post_data:
|
|
post_data = {"SAMLResponse": query["SAMLResponse"]} if "SAMLResponse" in query else {}
|
|
req_data = {
|
|
"https": https,
|
|
"http_host": request.headers.get("host", "localhost:8080"),
|
|
"script_name": "/auth/saml/logout",
|
|
"request_uri": request.url.path,
|
|
"query_string": str(request.url.query or ""),
|
|
"get_data": dict(query),
|
|
"post_data": post_data,
|
|
}
|
|
auth = OneLogin_Saml2_Auth(req_data, old_settings=settings)
|
|
try:
|
|
url = auth.process_slo(keep_local_session=True)
|
|
except Exception as err:
|
|
raise SAMLError(f"SAML logout could not be processed: {err}") from err
|
|
return url, auth.get_errors()
|