From 391b0db59827a9309cc61d91cd7166cd8d3aa37d Mon Sep 17 00:00:00 2001 From: miki3421 Date: Wed, 1 Jul 2026 16:36:43 +0000 Subject: [PATCH] Add Keycloak OIDC auth bridge --- openserverless/common/oidc_validator.py | 168 ++++++++++ openserverless/common/sso_namespace.py | 91 ++++++ openserverless/impl/auth/auth_service.py | 231 ++++++++++++- .../impl/auth/oidc_device_flow_service.py | 180 ++++++++++ openserverless/rest/auth.py | 155 ++++++++- tests/test_auth_service_oidc.py | 307 ++++++++++++++++++ tests/test_oidc_device_flow_service.py | 245 ++++++++++++++ tests/test_oidc_validator.py | 120 +++++++ tests/test_sso_namespace.py | 77 +++++ 9 files changed, 1567 insertions(+), 7 deletions(-) create mode 100644 openserverless/common/oidc_validator.py create mode 100644 openserverless/common/sso_namespace.py create mode 100644 openserverless/impl/auth/oidc_device_flow_service.py create mode 100644 tests/test_auth_service_oidc.py create mode 100644 tests/test_oidc_device_flow_service.py create mode 100644 tests/test_oidc_validator.py create mode 100644 tests/test_sso_namespace.py diff --git a/openserverless/common/oidc_validator.py b/openserverless/common/oidc_validator.py new file mode 100644 index 0000000..f47206d --- /dev/null +++ b/openserverless/common/oidc_validator.py @@ -0,0 +1,168 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import base64 +import json +import time + +import requests +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import padding, rsa + + +class OidcValidationError(Exception): + pass + + +class OidcForbiddenError(Exception): + pass + + +def _b64url_decode(value): + value += "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(value.encode("ascii")) + + +def _json_b64url_decode(value): + return json.loads(_b64url_decode(value)) + + +def _int_b64url_decode(value): + return int.from_bytes(_b64url_decode(value), byteorder="big") + + +class OidcTokenValidator: + + def __init__(self, environ, jwks=None, now=None): + self._environ = environ + self._jwks = jwks + self._now = now + + def _get_required(self, key): + value = self._environ.get(key) + if not value: + raise OidcValidationError(f"missing OIDC configuration: {key}") + return value + + def _get_jwks(self): + if self._jwks is not None: + return self._jwks + + jwks_url = self._get_required("OIDC_JWKS_URL") + response = requests.get(jwks_url, timeout=10) + response.raise_for_status() + self._jwks = response.json() + return self._jwks + + def _find_jwk(self, kid): + for jwk in self._get_jwks().get("keys", []): + if jwk.get("kid") == kid: + return jwk + raise OidcValidationError("no matching JWK found for token kid") + + def _public_key(self, jwk): + if jwk.get("kty") != "RSA": + raise OidcValidationError("unsupported JWK key type") + + public_numbers = rsa.RSAPublicNumbers( + e=_int_b64url_decode(jwk["e"]), + n=_int_b64url_decode(jwk["n"]), + ) + return public_numbers.public_key() + + def _verify_signature(self, token, header): + parts = token.split(".") + signed_data = f"{parts[0]}.{parts[1]}".encode("ascii") + signature = _b64url_decode(parts[2]) + jwk = self._find_jwk(header.get("kid")) + public_key = self._public_key(jwk) + + try: + public_key.verify( + signature, + signed_data, + padding.PKCS1v15(), + hashes.SHA256(), + ) + except Exception as exc: + raise OidcValidationError("invalid token signature") from exc + + def _validate_time_claims(self, claims): + now = self._now if self._now is not None else int(time.time()) + leeway = int(self._environ.get("OIDC_CLOCK_LEEWAY_SECONDS", "30")) + + exp = claims.get("exp") + if exp is None or int(exp) < now - leeway: + raise OidcValidationError("token expired") + + nbf = claims.get("nbf") + if nbf is not None and int(nbf) > now + leeway: + raise OidcValidationError("token is not valid yet") + + def _validate_issuer(self, claims): + expected = self._get_required("OIDC_ISSUER_URL") + if claims.get("iss") != expected: + raise OidcValidationError("invalid issuer") + + def _validate_audience(self, claims): + expected = self._get_required("OIDC_AUDIENCE") + audience = claims.get("aud") + if isinstance(audience, list) and expected in audience: + return + if audience == expected: + return + raise OidcValidationError("invalid audience") + + def _validate_required_group(self, claims): + required_group = self._environ.get("OIDC_REQUIRED_GROUP") + if not required_group: + return + + groups_claim = self._environ.get("OIDC_GROUPS_CLAIM", "groups") + groups = claims.get(groups_claim, []) + if isinstance(groups, str): + groups = [groups] + + if required_group not in groups: + raise OidcForbiddenError("missing required group") + + def validate(self, token): + if not token: + raise OidcValidationError("missing access token") + + parts = token.split(".") + if len(parts) != 3: + raise OidcValidationError("invalid token format") + + header = _json_b64url_decode(parts[0]) + claims = _json_b64url_decode(parts[1]) + + if header.get("alg") != "RS256": + raise OidcValidationError("unsupported token algorithm") + + self._verify_signature(token, header) + self._validate_issuer(claims) + self._validate_audience(claims) + self._validate_time_claims(claims) + self._validate_required_group(claims) + + username_claim = self._environ.get("OIDC_USERNAME_CLAIM", "preferred_username") + if not claims.get(username_claim): + raise OidcValidationError(f"missing username claim: {username_claim}") + + return claims + diff --git a/openserverless/common/sso_namespace.py b/openserverless/common/sso_namespace.py new file mode 100644 index 0000000..9153c5b --- /dev/null +++ b/openserverless/common/sso_namespace.py @@ -0,0 +1,91 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import hashlib +import re + +import openserverless.common.validation as validation + + +class SsoNamespaceMapper: + + def __init__(self, environ): + self._environ = environ + + def namespace_for(self, claims): + username_claim = self._environ.get("OIDC_USERNAME_CLAIM", "preferred_username") + external_username = claims[username_claim] + + namespace_claim = self._environ.get("OIDC_NAMESPACE_CLAIM") + if namespace_claim and claims.get(namespace_claim): + namespace = claims[namespace_claim] + if validation.is_valid_username(namespace): + return namespace + + if self._preserve_valid() and validation.is_valid_username(external_username): + return external_username + + return self._normalized_namespace(external_username, claims) + + def _normalized_namespace(self, external_username, claims): + base = re.sub(r"[^a-z0-9]", "", external_username.lower()) + if len(base) < 5: + base = f"{base}user" + + suffix = self._suffix(external_username, claims) + max_len = self._max_len() + prefix_len = max_len - len(suffix) + namespace = f"{base[:prefix_len]}{suffix}" + + if validation.is_valid_username(namespace): + return namespace + + # Defensive fallback for unusual inputs where the base becomes empty. + namespace = f"user{suffix}" + if validation.is_valid_username(namespace): + return namespace + + raise ValueError("Unable to derive a valid namespace from SSO claims") + + def _suffix(self, external_username, claims): + suffix_len = self._suffix_len() + source = "|".join( + [ + claims.get("iss", ""), + claims.get("sub", ""), + external_username, + ] + ) + return hashlib.sha256(source.encode("utf-8")).hexdigest()[:suffix_len] + + def _preserve_valid(self): + value = self._environ.get("SSO_NAMESPACE_PRESERVE_VALID", "true") + return str(value).lower() in ["1", "true", "yes", "on"] + + def _suffix_len(self): + try: + value = int(self._environ.get("SSO_NAMESPACE_HASH_LENGTH", 8)) + except (TypeError, ValueError): + value = 8 + return min(max(value, 6), 16) + + def _max_len(self): + try: + value = int(self._environ.get("SSO_NAMESPACE_MAX_LENGTH", 61)) + except (TypeError, ValueError): + value = 61 + return min(max(value, 13), 61) diff --git a/openserverless/impl/auth/auth_service.py b/openserverless/impl/auth/auth_service.py index d13fde7..9a62f6f 100644 --- a/openserverless/impl/auth/auth_service.py +++ b/openserverless/impl/auth/auth_service.py @@ -19,21 +19,36 @@ import json import os import logging +import secrets +import string +import time import openserverless.common.response_builder as res_builder import openserverless.couchdb.bcrypt_util as bu +from openserverless.common.oidc_validator import ( + OidcForbiddenError, + OidcTokenValidator, + OidcValidationError, +) +from openserverless.common.sso_namespace import SsoNamespaceMapper from openserverless.couchdb.couchdb_util import CouchDB from openserverless.common.kube_api_client import KubeApiClient USER_META_DBN = "users_metadata" +SSO_PROVIDER_ANNOTATION = "openserverless.apache.org/auth-provider" +SSO_MODE_ANNOTATION = "openserverless.apache.org/auth-mode" +SSO_USERNAME_ANNOTATION = "openserverless.apache.org/sso-username" +SSO_SUB_ANNOTATION = "openserverless.apache.org/sso-sub" +SSO_ISSUER_ANNOTATION = "openserverless.apache.org/sso-issuer" +SSO_DISABLED_ANNOTATION = "openserverless.apache.org/sso-disabled" class AuthService: - def __init__(self, environ=os.environ): + def __init__(self, environ=os.environ, couch_db=None, kube_client=None): self._environ = environ - self.couch_db = CouchDB() - self.kube_client = KubeApiClient() + self.couch_db = couch_db if couch_db is not None else CouchDB() + self.kube_client = kube_client if kube_client is not None else KubeApiClient() def fetch_user_data(self, login: str): logging.info(f"searching for user {login} data") @@ -53,7 +68,7 @@ def fetch_user_data(self, login: str): f"failed to query OpenServerless metadata for user {login}. Reason: {e}" ) return None - + def env_to_dict(self, user_data, key="env"): """ extract env from user_data and return it as a dict @@ -101,8 +116,212 @@ def login(self, login, password): else: logging.warning(f"no user {login} found") return res_builder.build_error_message(f"Invalid credentials", 401) - - + + def login_oidc(self, access_token, expected_namespace=None): + try: + validator = OidcTokenValidator(self._environ) + claims = validator.validate(access_token) + except OidcForbiddenError: + return res_builder.build_error_message("Forbidden", 403) + except OidcValidationError as exc: + logging.warning(f"OIDC token validation failed: {exc}") + return res_builder.build_error_message("Invalid OIDC token", 401) + + username_claim = self._environ.get("OIDC_USERNAME_CLAIM", "preferred_username") + external_username = claims[username_claim] + try: + login = SsoNamespaceMapper(self._environ).namespace_for(claims) + except ValueError as exc: + logging.warning(f"OIDC user {external_username} namespace mapping failed: {exc}") + return res_builder.build_error_message("Invalid OIDC namespace mapping", 400) + + if expected_namespace and login != expected_namespace: + logging.warning( + f"OIDC user {external_username} resolved namespace {login} " + f"does not match requested namespace {expected_namespace}" + ) + return res_builder.build_error_message( + "Authenticated namespace does not match requested workspace", + 403, + ) + + if self.is_sso_login_disabled(login): + logging.warning(f"OIDC user {external_username} is unbound from namespace {login}") + return res_builder.build_error_message("Forbidden", 403) + + user_data = self.fetch_user_data(login) + + if not user_data: + user_data = self.provision_oidc_user_if_enabled(login, external_username, claims) + + if not user_data: + logging.warning(f"OIDC user {external_username} has no provisioned namespace") + return res_builder.build_error_message("Namespace not provisioned", 404) + + login_data = self.map_login_data(user_data) + if "AUTH" not in login_data: + logging.error(f"OIDC user {login} metadata is missing AUTH") + return res_builder.build_error_message("Namespace metadata is missing AUTH", 500) + + login_data["LOGIN"] = login + login_data["NAMESPACE"] = login + return res_builder.build_response_with_data(login_data) + + def provision_oidc_user_if_enabled(self, login, external_username, claims): + if not self._is_truthy(self._environ.get("SSO_AUTOPROVISION_ON_LOGIN")): + return None + + email = claims.get("email") + if not email: + logging.warning(f"OIDC user {login} cannot be provisioned without email") + return None + + if not self.ensure_whisk_user(login, external_username, email, claims): + return None + + return self.wait_for_user_data(login) + + def ensure_whisk_user(self, login, external_username, email, claims): + existing_whisk_user = self.kube_client.get_whisk_user(login) + if existing_whisk_user: + logging.info(f"WhiskUser for OIDC user {login} already exists") + return True + + whisk_user = self.build_sso_whisk_user(login, external_username, email, claims) + if self.kube_client.create_whisk_user(whisk_user): + logging.info(f"WhiskUser for OIDC user {login} created") + return True + + # A concurrent login may have created the CR between get and create. + if self.kube_client.get_whisk_user(login): + logging.info(f"WhiskUser for OIDC user {login} found after create conflict") + return True + + logging.error(f"failed to create WhiskUser for OIDC user {login}") + return False + + def build_sso_whisk_user(self, login, external_username, email, claims): + auth = self._random_auth() + password = self._random_secret(24) + services = self._environ.get("SSO_AUTOPROVISION_DEFAULT_SERVICES", "all") + + whisk_user = { + "apiVersion": "nuvolaris.org/v1", + "kind": "WhiskUser", + "metadata": { + "name": login, + "namespace": "nuvolaris", + "annotations": { + SSO_PROVIDER_ANNOTATION: self._environ.get("OIDC_PROVIDER", "keycloak"), + SSO_MODE_ANNOTATION: "sso", + SSO_USERNAME_ANNOTATION: external_username, + SSO_SUB_ANNOTATION: claims.get("sub", ""), + SSO_ISSUER_ANNOTATION: claims.get("iss", ""), + }, + }, + "spec": { + "email": email, + "password": password, + "namespace": login, + "auth": auth, + }, + } + + if services == "all": + whisk_user["spec"]["redis"] = { + "enabled": True, + "prefix": login, + "password": self._random_secret(12), + } + whisk_user["spec"]["mongodb"] = { + "enabled": True, + "database": login, + "password": self._random_secret(12), + } + whisk_user["spec"]["postgres"] = { + "enabled": True, + "database": login, + "password": self._random_secret(12), + } + whisk_user["spec"]["object-storage"] = { + "password": self._random_secret(40), + "quota": self._environ.get("SSO_AUTOPROVISION_STORAGE_QUOTA", "auto"), + "data": {"enabled": True, "bucket": f"{login}-data"}, + "route": {"enabled": True, "bucket": f"{login}-web"}, + } + whisk_user["spec"]["milvus"] = { + "enabled": True, + "database": login, + "password": self._random_secret(12), + } + + return whisk_user + + def wait_for_user_data(self, login): + timeout_seconds = self._int_env("SSO_AUTOPROVISION_TIMEOUT_SECONDS", 120) + poll_seconds = self._float_env("SSO_AUTOPROVISION_POLL_SECONDS", 2) + deadline = time.monotonic() + timeout_seconds + + while True: + user_data = self.fetch_user_data(login) + if user_data: + return user_data + + if time.monotonic() >= deadline: + logging.warning(f"timeout waiting for OIDC namespace {login}") + return None + + time.sleep(poll_seconds) + + def _random_auth(self): + return f"{self._random_uuid()}:{self._random_secret(64)}" + + def _random_uuid(self): + import uuid + return str(uuid.uuid4()) + + def _random_secret(self, length): + alphabet = string.ascii_letters + string.digits + return "".join(secrets.choice(alphabet) for _ in range(length)) + + def _is_truthy(self, value): + return str(value or "").lower() in ["1", "true", "yes", "on"] + + def _int_env(self, name, default): + try: + return int(self._environ.get(name, default)) + except (TypeError, ValueError): + return default + + def _float_env(self, name, default): + try: + return float(self._environ.get(name, default)) + except (TypeError, ValueError): + return default + + def is_sso_login_disabled(self, login): + if not hasattr(self.kube_client, "get_whisk_user"): + return False + + whisk_user = self.kube_client.get_whisk_user(login) + annotations = ((whisk_user or {}).get("metadata") or {}).get("annotations") or {} + return self._is_truthy(annotations.get(SSO_DISABLED_ANNOTATION)) + + def map_login_data(self, user_data): + """ + Map user metadata to the flat string-only shape consumed by ops -login. + This mirrors the whisk-system/nuv/login system action response. + """ + body = self.env_to_dict(user_data, "userenv") + + for key, value in self.env_to_dict(user_data, "env").items(): + if key not in body: + body[key] = value + else: + body[f"SYSTEM_{key}"] = value + + return body + def update_password(self, login, old_password, new_password): user_data = self.fetch_user_data(login) diff --git a/openserverless/impl/auth/oidc_device_flow_service.py b/openserverless/impl/auth/oidc_device_flow_service.py new file mode 100644 index 0000000..fcad448 --- /dev/null +++ b/openserverless/impl/auth/oidc_device_flow_service.py @@ -0,0 +1,180 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import base64 +import hashlib +import os +import secrets +import time + +import requests + +import openserverless.common.response_builder as res_builder +from openserverless.impl.auth.auth_service import AuthService + + +_DEVICE_FLOWS = {} + + +class OidcDeviceFlowService: + + def __init__( + self, + environ=os.environ, + http_client=requests, + auth_service=None, + store=None, + now=None, + ): + self._environ = environ + self._http_client = http_client + self._auth_service = auth_service if auth_service is not None else AuthService(environ=environ) + self._store = store if store is not None else _DEVICE_FLOWS + self._now = now if now is not None else time.time + + def start(self, requested_namespace=None): + try: + verifier, challenge = self._create_pkce_pair() + response = self._http_client.post( + self._device_authorization_url(), + data={ + "client_id": self._client_id(), + "scope": self._environ.get("OIDC_DEVICE_SCOPE", "openid email profile"), + "code_challenge": challenge, + "code_challenge_method": "S256", + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=10, + ) + payload = response.json() + except Exception: + return res_builder.build_error_message("Unable to start SSO login", 502) + + if response.status_code >= 400: + return res_builder.build_error_message( + payload.get("error_description") or payload.get("error") or "Unable to start SSO login", + 502, + ) + + flow_id = secrets.token_urlsafe(32) + expires_in = int(payload.get("expires_in", 600)) + self._cleanup_expired() + self._store[flow_id] = { + "device_code": payload["device_code"], + "verifier": verifier, + "expires_at": self._now() + expires_in, + "interval": int(payload.get("interval", 5)), + "requested_namespace": requested_namespace, + } + + return res_builder.build_response_with_data( + { + "flow_id": flow_id, + "user_code": payload.get("user_code"), + "verification_uri": payload.get("verification_uri"), + "verification_uri_complete": payload.get("verification_uri_complete"), + "expires_in": expires_in, + "interval": int(payload.get("interval", 5)), + } + ) + + def poll(self, flow_id): + flow = self._store.get(flow_id) + if not flow: + return res_builder.build_error_message("Invalid SSO flow", 404) + + if self._now() >= flow["expires_at"]: + self._store.pop(flow_id, None) + return res_builder.build_error_message("SSO login expired", 400) + + try: + response = self._http_client.post( + self._token_url(), + data={ + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + "client_id": self._client_id(), + "device_code": flow["device_code"], + "code_verifier": flow["verifier"], + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=10, + ) + payload = response.json() + except Exception: + return res_builder.build_error_message("Unable to poll SSO login", 502) + + error = payload.get("error") + if error in ("authorization_pending", "slow_down"): + return res_builder.build_response_with_data( + { + "status": "pending", + "message": error, + "interval": flow["interval"], + }, + 202, + ) + + if response.status_code >= 400: + self._store.pop(flow_id, None) + return res_builder.build_error_message( + payload.get("error_description") or error or "SSO login failed", + 401, + ) + + access_token = payload.get("access_token") + if not access_token: + self._store.pop(flow_id, None) + return res_builder.build_error_message("SSO login did not return an access token", 401) + + self._store.pop(flow_id, None) + return self._auth_service.login_oidc( + access_token, + expected_namespace=flow.get("requested_namespace"), + ) + + def _client_id(self): + return self._environ.get("OIDC_CLIENT_ID") or self._environ.get("OIDC_AUDIENCE") + + def _device_authorization_url(self): + return self._environ.get("OIDC_DEVICE_AUTHORIZATION_URL") or ( + f"{self._issuer_url()}/protocol/openid-connect/auth/device" + ) + + def _token_url(self): + return self._environ.get("OIDC_TOKEN_URL") or ( + f"{self._issuer_url()}/protocol/openid-connect/token" + ) + + def _issuer_url(self): + issuer = self._environ.get("OIDC_ISSUER_URL") + if not issuer: + raise ValueError("missing OIDC_ISSUER_URL") + return issuer.rstrip("/") + + def _cleanup_expired(self): + now = self._now() + expired = [flow_id for flow_id, flow in self._store.items() if now >= flow["expires_at"]] + for flow_id in expired: + self._store.pop(flow_id, None) + + def _create_pkce_pair(self): + verifier = self._base64_url(secrets.token_bytes(32)) + challenge = self._base64_url(hashlib.sha256(verifier.encode("ascii")).digest()) + return verifier, challenge + + def _base64_url(self, value): + return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") diff --git a/openserverless/rest/auth.py b/openserverless/rest/auth.py index d6bc4e1..90492ce 100644 --- a/openserverless/rest/auth.py +++ b/openserverless/rest/auth.py @@ -19,10 +19,41 @@ from openserverless import app from openserverless.impl.auth.auth_service import AuthService +from openserverless.impl.auth.oidc_device_flow_service import OidcDeviceFlowService from openserverless.security.ow_authorize import ow_authorize from flask import request import openserverless.common.response_builder as res_builder from flasgger import swag_from +from urllib.parse import urlparse + + +def _extract_bearer_token(): + authorization = request.headers.get("Authorization", "") + if authorization.lower().startswith("bearer "): + return authorization.split(" ", 1)[1].strip() + + body = request.get_json(silent=True) or {} + return body.get("access_token") + + +def _requested_namespace_from_origin(origin, api_host): + if not origin: + return None + + origin_host = urlparse(origin).hostname + if not origin_host: + return None + + api_hostname = (api_host or "").split(":", 1)[0] + suffix = f".{api_hostname}" + if origin_host == api_hostname or not origin_host.endswith(suffix): + return None + + candidate = origin_host[: -len(suffix)] + if "." in candidate or not candidate: + return None + + return candidate @app.route('/system/api/v1/auth/',methods=['PATCH']) @ow_authorize(pass_user_data=True) @@ -111,4 +142,126 @@ def login(): login_data = request.get_json() auth_service = AuthService() return auth_service.login(login_data['login'], login_data['password']) - \ No newline at end of file + +@app.route('/system/api/v1/auth/oidc',methods=['POST']) +def login_oidc(): + """ + User Authentication with OIDC + --- + tags: + - Authentication Api + summary: Authenticate user with an OIDC access token + description: Validate a Keycloak OIDC token and map it to an existing OpenServerless namespace. + operationId: authenticateUserOidc + consumes: + - application/json + parameters: + - in: header + name: Authorization + description: Bearer token issued by the configured OIDC provider + required: false + type: string + - in: body + name: OidcCredentials + description: Optional JSON payload containing access_token when Authorization header is not used + required: false + schema: + type: object + properties: + access_token: + type: string + responses: + 200: + description: Authentication successful. Returns OpenServerless user data. + schema: + $ref: '#/definitions/MessageData' + 401: + description: Unauthorized. Invalid or missing OIDC token. + schema: + $ref: '#/definitions/Message' + 403: + description: Forbidden. Token is valid but required group is missing. + schema: + $ref: '#/definitions/Message' + 404: + description: OIDC user is valid but no OpenServerless namespace is provisioned. + schema: + $ref: '#/definitions/Message' + """ + auth_service = AuthService() + return auth_service.login_oidc(_extract_bearer_token()) + + +@app.route('/system/api/v1/auth/oidc/device/start', methods=['POST']) +def start_oidc_device_login(): + """ + Start backend-managed OIDC Device Authorization login + --- + tags: + - Authentication Api + summary: Start an SSO login flow managed by admin-api + description: admin-api uses cluster SSO configuration to start OIDC device login and returns only an opaque flow id and user verification data. + operationId: startOidcDeviceLogin + responses: + 200: + description: SSO device login flow started. + schema: + type: object + 502: + description: admin-api could not start login with the configured OIDC provider. + schema: + $ref: '#/definitions/Message' + """ + return OidcDeviceFlowService().start( + requested_namespace=_requested_namespace_from_origin( + request.headers.get("Origin"), + request.host, + ) + ) + + +@app.route('/system/api/v1/auth/oidc/device/poll', methods=['POST']) +def poll_oidc_device_login(): + """ + Poll backend-managed OIDC Device Authorization login + --- + tags: + - Authentication Api + summary: Poll an SSO login flow managed by admin-api + description: admin-api exchanges device credentials with the configured OIDC provider, validates the token and returns OpenServerless namespace data. OIDC tokens are not returned to the browser. + operationId: pollOidcDeviceLogin + consumes: + - application/json + parameters: + - in: body + name: OidcDevicePoll + required: true + schema: + type: object + properties: + flow_id: + type: string + responses: + 200: + description: Authentication successful. Returns OpenServerless user data. + schema: + $ref: '#/definitions/MessageData' + 202: + description: Login is still pending at the identity provider. + schema: + type: object + 400: + description: Login flow expired. + schema: + $ref: '#/definitions/Message' + 401: + description: SSO login failed. + schema: + $ref: '#/definitions/Message' + 404: + description: Unknown login flow or namespace not provisioned. + schema: + $ref: '#/definitions/Message' + """ + body = request.get_json(silent=True) or {} + return OidcDeviceFlowService().poll(body.get("flow_id")) diff --git a/tests/test_auth_service_oidc.py b/tests/test_auth_service_oidc.py new file mode 100644 index 0000000..ebdcb3d --- /dev/null +++ b/tests/test_auth_service_oidc.py @@ -0,0 +1,307 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import unittest +from unittest.mock import patch + +from openserverless import app +from openserverless.impl.auth.auth_service import AuthService + + +class FakeCouchDB: + + def __init__(self, docs): + self.docs = docs + + def find_doc(self, db_name, selector): + return {"docs": self.docs} + + +class SequencedCouchDB: + + def __init__(self, docs_by_call): + self.docs_by_call = list(docs_by_call) + self.calls = 0 + + def find_doc(self, db_name, selector): + index = min(self.calls, len(self.docs_by_call) - 1) + self.calls += 1 + return {"docs": self.docs_by_call[index]} + + +class FakeKubeClient: + + def __init__(self, existing=None, create_result=True): + self.existing = existing + self.create_result = create_result + self.created = [] + + def get_whisk_user(self, username): + return self.existing + + def create_whisk_user(self, whisk_user): + self.created.append(whisk_user) + return self.create_result + + +class AuthServiceOidcTest(unittest.TestCase): + + def service(self, docs, environ=None, kube_client=None): + env = { + "OIDC_USERNAME_CLAIM": "preferred_username", + } + if environ: + env.update(environ) + return AuthService( + environ=env, + couch_db=FakeCouchDB(docs), + kube_client=kube_client if kube_client is not None else object(), + ) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_returns_ops_login_data(self, validator_class): + validator_class.return_value.validate.return_value = { + "preferred_username": "devel", + } + service = self.service( + [ + { + "login": "devel", + "email": "devel@example.test", + "env": [ + {"key": "AUTH", "value": "uuid:key"}, + {"key": "APIHOST", "value": "https://ops.example.test"}, + {"key": "SHARED", "value": "system-value"}, + ], + "userenv": [ + {"key": "SHARED", "value": "user-value"}, + ], + } + ] + ) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(200, response.status_code) + self.assertEqual("uuid:key", response.json["AUTH"]) + self.assertEqual("https://ops.example.test", response.json["APIHOST"]) + self.assertEqual("devel", response.json["LOGIN"]) + self.assertEqual("devel", response.json["NAMESPACE"]) + self.assertEqual("user-value", response.json["SHARED"]) + self.assertEqual("system-value", response.json["SYSTEM_SHARED"]) + self.assertNotIn("quota", response.json) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_returns_404_when_namespace_is_missing(self, validator_class): + validator_class.return_value.validate.return_value = { + "preferred_username": "missing.lab", + } + service = self.service([]) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(404, response.status_code) + self.assertEqual("ko", response.json["status"]) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_returns_403_when_sso_binding_is_disabled(self, validator_class): + validator_class.return_value.validate.return_value = { + "preferred_username": "devel", + } + service = self.service( + [ + { + "login": "devel", + "email": "devel@example.test", + "env": [{"key": "AUTH", "value": "uuid:key"}], + } + ], + kube_client=FakeKubeClient( + existing={ + "metadata": { + "annotations": { + "openserverless.apache.org/sso-disabled": "true", + }, + }, + }, + ), + ) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(403, response.status_code) + self.assertEqual("ko", response.json["status"]) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_autoprovisions_whisk_user_when_enabled(self, validator_class): + validator_class.return_value.validate.return_value = { + "iss": "http://issuer.test", + "sub": "keycloak-subject", + "preferred_username": "ssouser", + "email": "sso.user@example.test", + } + kube_client = FakeKubeClient() + service = AuthService( + environ={ + "OIDC_USERNAME_CLAIM": "preferred_username", + "SSO_AUTOPROVISION_ON_LOGIN": "true", + "SSO_AUTOPROVISION_TIMEOUT_SECONDS": "1", + "SSO_AUTOPROVISION_POLL_SECONDS": "0", + }, + couch_db=SequencedCouchDB( + [ + [], + [ + { + "login": "ssouser", + "email": "sso.user@example.test", + "env": [ + {"key": "AUTH", "value": "uuid:key"}, + {"key": "APIHOST", "value": "https://ops.example.test"}, + ], + } + ], + ] + ), + kube_client=kube_client, + ) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(200, response.status_code) + self.assertEqual("uuid:key", response.json["AUTH"]) + self.assertEqual("ssouser", response.json["LOGIN"]) + self.assertEqual("ssouser", response.json["NAMESPACE"]) + self.assertEqual(1, len(kube_client.created)) + + whisk_user = kube_client.created[0] + self.assertEqual("WhiskUser", whisk_user["kind"]) + self.assertEqual("ssouser", whisk_user["metadata"]["name"]) + self.assertEqual("ssouser", whisk_user["spec"]["namespace"]) + self.assertEqual("sso.user@example.test", whisk_user["spec"]["email"]) + self.assertIn(":", whisk_user["spec"]["auth"]) + self.assertNotEqual("", whisk_user["spec"]["password"]) + self.assertEqual( + "keycloak", + whisk_user["metadata"]["annotations"][ + "openserverless.apache.org/auth-provider" + ], + ) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_rejects_namespace_mismatch_before_autoprovision(self, validator_class): + validator_class.return_value.validate.return_value = { + "iss": "http://issuer.test", + "sub": "developer-subject", + "preferred_username": "developer.lab", + "email": "developer.lab@example.test", + } + kube_client = FakeKubeClient() + service = AuthService( + environ={ + "OIDC_USERNAME_CLAIM": "preferred_username", + "SSO_AUTOPROVISION_ON_LOGIN": "true", + }, + couch_db=FakeCouchDB([]), + kube_client=kube_client, + ) + + with app.app_context(): + response = service.login_oidc("token", expected_namespace="michelem") + + self.assertEqual(403, response.status_code) + self.assertEqual("ko", response.json["status"]) + self.assertEqual( + "Authenticated namespace does not match requested workspace", + response.json["message"], + ) + self.assertEqual([], kube_client.created) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_autoprovisions_normalized_namespace(self, validator_class): + validator_class.return_value.validate.return_value = { + "iss": "http://issuer.test", + "sub": "developer-subject", + "preferred_username": "developer.lab", + "email": "developer.lab@example.test", + } + kube_client = FakeKubeClient() + service = AuthService( + environ={ + "OIDC_USERNAME_CLAIM": "preferred_username", + "SSO_AUTOPROVISION_ON_LOGIN": "true", + "SSO_AUTOPROVISION_TIMEOUT_SECONDS": "1", + "SSO_AUTOPROVISION_POLL_SECONDS": "0", + }, + couch_db=SequencedCouchDB( + [ + [], + [ + { + "login": "developerlab4717c17e", + "email": "developer.lab@example.test", + "env": [{"key": "AUTH", "value": "uuid:key"}], + } + ], + ] + ), + kube_client=kube_client, + ) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(200, response.status_code) + self.assertEqual("developerlab4717c17e", response.json["LOGIN"]) + self.assertEqual("developerlab4717c17e", response.json["NAMESPACE"]) + self.assertEqual(1, len(kube_client.created)) + self.assertEqual("developerlab4717c17e", kube_client.created[0]["metadata"]["name"]) + self.assertEqual( + "developer.lab", + kube_client.created[0]["metadata"]["annotations"][ + "openserverless.apache.org/sso-username" + ], + ) + + @patch("openserverless.impl.auth.auth_service.OidcTokenValidator") + def test_oidc_login_returns_500_when_auth_is_missing(self, validator_class): + validator_class.return_value.validate.return_value = { + "preferred_username": "developer.lab", + } + service = self.service( + [ + { + "login": "developer.lab", + "email": "developer.lab@example.test", + "env": [{"key": "APIHOST", "value": "https://ops.example.test"}], + } + ] + ) + + with app.app_context(): + response = service.login_oidc("token") + + self.assertEqual(500, response.status_code) + self.assertEqual("ko", response.json["status"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_oidc_device_flow_service.py b/tests/test_oidc_device_flow_service.py new file mode 100644 index 0000000..749d6ca --- /dev/null +++ b/tests/test_oidc_device_flow_service.py @@ -0,0 +1,245 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import unittest + +import openserverless.common.response_builder as res_builder +from openserverless import app +from openserverless.impl.auth.oidc_device_flow_service import OidcDeviceFlowService + + +class FakeResponse: + + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + + def json(self): + return self._payload + + +class FakeHttpClient: + + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + + def post(self, url, data=None, headers=None, timeout=None): + self.calls.append( + { + "url": url, + "data": data, + "headers": headers, + "timeout": timeout, + } + ) + return self.responses.pop(0) + + +class FakeAuthService: + + def __init__(self): + self.tokens = [] + + def login_oidc(self, access_token, expected_namespace=None): + self.tokens.append( + { + "access_token": access_token, + "expected_namespace": expected_namespace, + } + ) + return res_builder.build_response_with_data( + { + "LOGIN": "michelem", + "NAMESPACE": "michelem", + "AUTH": "uuid:key", + } + ) + + +class OidcDeviceFlowServiceTest(unittest.TestCase): + + def setUp(self): + self.environ = { + "OIDC_ISSUER_URL": "https://keycloak.example.test/realms/openserverless-lab", + "OIDC_AUDIENCE": "openserverless-admin-api", + } + + def test_start_creates_backend_flow_without_returning_oidc_details(self): + store = {} + http_client = FakeHttpClient( + [ + FakeResponse( + { + "device_code": "device-secret", + "user_code": "ABCD-EFGH", + "verification_uri": "https://keycloak.example.test/device", + "verification_uri_complete": "https://keycloak.example.test/device?user_code=ABCD-EFGH", + "expires_in": 600, + "interval": 5, + } + ) + ] + ) + service = OidcDeviceFlowService( + environ=self.environ, + http_client=http_client, + auth_service=FakeAuthService(), + store=store, + now=lambda: 1000, + ) + + with app.app_context(): + response = service.start(requested_namespace="michelem") + + self.assertEqual(200, response.status_code) + self.assertEqual("ABCD-EFGH", response.json["user_code"]) + self.assertEqual("https://keycloak.example.test/device?user_code=ABCD-EFGH", response.json["verification_uri_complete"]) + self.assertIn("flow_id", response.json) + self.assertNotIn("device_code", response.json) + self.assertNotIn("code_verifier", response.json) + self.assertEqual(1, len(store)) + self.assertEqual("device-secret", next(iter(store.values()))["device_code"]) + self.assertEqual("michelem", next(iter(store.values()))["requested_namespace"]) + + start_call = http_client.calls[0] + self.assertEqual( + "https://keycloak.example.test/realms/openserverless-lab/protocol/openid-connect/auth/device", + start_call["url"], + ) + self.assertEqual("openserverless-admin-api", start_call["data"]["client_id"]) + self.assertEqual("S256", start_call["data"]["code_challenge_method"]) + + def test_poll_returns_pending_without_exposing_token(self): + store = { + "flow-1": { + "device_code": "device-secret", + "verifier": "verifier", + "expires_at": 2000, + "interval": 5, + } + } + http_client = FakeHttpClient([FakeResponse({"error": "authorization_pending"}, 400)]) + service = OidcDeviceFlowService( + environ=self.environ, + http_client=http_client, + auth_service=FakeAuthService(), + store=store, + now=lambda: 1000, + ) + + with app.app_context(): + response = service.poll("flow-1") + + self.assertEqual(202, response.status_code) + self.assertEqual("pending", response.json["status"]) + self.assertEqual("authorization_pending", response.json["message"]) + self.assertIn("flow-1", store) + + def test_poll_exchanges_token_and_returns_openserverless_login(self): + store = { + "flow-1": { + "device_code": "device-secret", + "verifier": "verifier", + "expires_at": 2000, + "interval": 5, + } + } + auth_service = FakeAuthService() + http_client = FakeHttpClient([FakeResponse({"access_token": "oidc-token"})]) + service = OidcDeviceFlowService( + environ=self.environ, + http_client=http_client, + auth_service=auth_service, + store=store, + now=lambda: 1000, + ) + + with app.app_context(): + response = service.poll("flow-1") + + self.assertEqual(200, response.status_code) + self.assertEqual("michelem", response.json["NAMESPACE"]) + self.assertEqual( + [{"access_token": "oidc-token", "expected_namespace": None}], + auth_service.tokens, + ) + self.assertNotIn("flow-1", store) + + token_call = http_client.calls[0] + self.assertEqual( + "https://keycloak.example.test/realms/openserverless-lab/protocol/openid-connect/token", + token_call["url"], + ) + self.assertEqual("device-secret", token_call["data"]["device_code"]) + self.assertEqual("verifier", token_call["data"]["code_verifier"]) + + def test_poll_passes_requested_namespace_to_oidc_login(self): + store = { + "flow-1": { + "device_code": "device-secret", + "verifier": "verifier", + "expires_at": 2000, + "interval": 5, + "requested_namespace": "michelem", + } + } + auth_service = FakeAuthService() + http_client = FakeHttpClient([FakeResponse({"access_token": "oidc-token"})]) + service = OidcDeviceFlowService( + environ=self.environ, + http_client=http_client, + auth_service=auth_service, + store=store, + now=lambda: 1000, + ) + + with app.app_context(): + response = service.poll("flow-1") + + self.assertEqual(200, response.status_code) + self.assertEqual( + [{"access_token": "oidc-token", "expected_namespace": "michelem"}], + auth_service.tokens, + ) + + def test_poll_expired_flow(self): + store = { + "flow-1": { + "device_code": "device-secret", + "verifier": "verifier", + "expires_at": 1000, + "interval": 5, + } + } + service = OidcDeviceFlowService( + environ=self.environ, + http_client=FakeHttpClient([]), + auth_service=FakeAuthService(), + store=store, + now=lambda: 1000, + ) + + with app.app_context(): + response = service.poll("flow-1") + + self.assertEqual(400, response.status_code) + self.assertNotIn("flow-1", store) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_oidc_validator.py b/tests/test_oidc_validator.py new file mode 100644 index 0000000..810f4c6 --- /dev/null +++ b/tests/test_oidc_validator.py @@ -0,0 +1,120 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import base64 +import json +import unittest + +from cryptography.hazmat.primitives.asymmetric import padding, rsa +from cryptography.hazmat.primitives import hashes + +from openserverless.common.oidc_validator import ( + OidcForbiddenError, + OidcTokenValidator, + OidcValidationError, +) + + +def b64url(data): + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def int_to_b64url(value): + length = (value.bit_length() + 7) // 8 + return b64url(value.to_bytes(length, byteorder="big")) + + +class OidcValidatorTest(unittest.TestCase): + + def setUp(self): + self.private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_numbers = self.private_key.public_key().public_numbers() + self.jwks = { + "keys": [ + { + "kty": "RSA", + "kid": "test-key", + "alg": "RS256", + "use": "sig", + "n": int_to_b64url(public_numbers.n), + "e": int_to_b64url(public_numbers.e), + } + ] + } + self.environ = { + "OIDC_ISSUER_URL": "http://localhost:8080/realms/openserverless-lab", + "OIDC_AUDIENCE": "openserverless-admin-api", + "OIDC_REQUIRED_GROUP": "openserverless-users", + "OIDC_USERNAME_CLAIM": "preferred_username", + "OIDC_GROUPS_CLAIM": "groups", + } + + def token(self, claims, header=None): + header = header or {"alg": "RS256", "kid": "test-key", "typ": "JWT"} + encoded_header = b64url(json.dumps(header).encode("utf-8")) + encoded_claims = b64url(json.dumps(claims).encode("utf-8")) + signed_data = f"{encoded_header}.{encoded_claims}".encode("ascii") + signature = self.private_key.sign( + signed_data, + padding.PKCS1v15(), + hashes.SHA256(), + ) + return f"{encoded_header}.{encoded_claims}.{b64url(signature)}" + + def valid_claims(self): + return { + "iss": self.environ["OIDC_ISSUER_URL"], + "aud": self.environ["OIDC_AUDIENCE"], + "exp": 2000, + "preferred_username": "developer.lab", + "email": "developer.lab@example.test", + "groups": ["openserverless-users"], + } + + def validator(self): + return OidcTokenValidator(self.environ, jwks=self.jwks, now=1000) + + def test_accepts_valid_token(self): + claims = self.validator().validate(self.token(self.valid_claims())) + + self.assertEqual("developer.lab", claims["preferred_username"]) + + def test_rejects_missing_required_group(self): + claims = self.valid_claims() + claims["groups"] = [] + + with self.assertRaises(OidcForbiddenError): + self.validator().validate(self.token(claims)) + + def test_rejects_wrong_audience(self): + claims = self.valid_claims() + claims["aud"] = "other-client" + + with self.assertRaises(OidcValidationError): + self.validator().validate(self.token(claims)) + + def test_rejects_expired_token(self): + claims = self.valid_claims() + claims["exp"] = 900 + + with self.assertRaises(OidcValidationError): + self.validator().validate(self.token(claims)) + + +if __name__ == "__main__": + unittest.main() + diff --git a/tests/test_sso_namespace.py b/tests/test_sso_namespace.py new file mode 100644 index 0000000..78eea6a --- /dev/null +++ b/tests/test_sso_namespace.py @@ -0,0 +1,77 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import unittest + +import openserverless.common.validation as validation +from openserverless.common.sso_namespace import SsoNamespaceMapper + + +class SsoNamespaceMapperTest(unittest.TestCase): + + def claims(self, preferred_username): + return { + "iss": "http://localhost:8080/realms/openserverless-lab", + "sub": "keycloak-subject", + "preferred_username": preferred_username, + } + + def test_preserves_valid_namespace(self): + namespace = SsoNamespaceMapper({}).namespace_for(self.claims("devel")) + + self.assertEqual("devel", namespace) + + def test_normalizes_invalid_username_with_hash_suffix(self): + namespace = SsoNamespaceMapper({}).namespace_for(self.claims("developer.lab")) + + self.assertTrue(namespace.startswith("developerlab")) + self.assertTrue(validation.is_valid_username(namespace)) + self.assertLessEqual(len(namespace), 61) + + def test_hash_suffix_keeps_colliding_bases_unique(self): + first = SsoNamespaceMapper({}).namespace_for( + { + "iss": "issuer", + "sub": "first", + "preferred_username": "developer.lab", + } + ) + second = SsoNamespaceMapper({}).namespace_for( + { + "iss": "issuer", + "sub": "second", + "preferred_username": "developer_lab", + } + ) + + self.assertNotEqual(first, second) + + def test_can_use_explicit_namespace_claim_when_valid(self): + namespace = SsoNamespaceMapper( + {"OIDC_NAMESPACE_CLAIM": "openserverless_namespace"} + ).namespace_for( + { + "preferred_username": "developer.lab", + "openserverless_namespace": "customns", + } + ) + + self.assertEqual("customns", namespace) + + +if __name__ == "__main__": + unittest.main()