diff --git a/.github/workflows/image.yml b/.github/workflows/image.yml index b13c865..016e0f5 100644 --- a/.github/workflows/image.yml +++ b/.github/workflows/image.yml @@ -95,7 +95,7 @@ jobs: echo "REGISTRY_TOKEN=${{ secrets.DOCKERHUB_TOKEN }}" >> "$GITHUB_ENV" fi - name: Registry login - uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0 + uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0 with: registry: ${{ env.REGISTRY }} username: ${{ env.REGISTRY_USER }} diff --git a/README.md b/README.md index 22ff510..3c11e54 100644 --- a/README.md +++ b/README.md @@ -148,4 +148,6 @@ https://github.com/apache/openserverless-admin-api/actions ## Additional Documentation -- [Deployer](docs/DEPLOYER.md) \ No newline at end of file +- [Deployer](docs/DEPLOYER.md) +- [External user provisioning](docs/USER_PROVISIONING.md): optional Secret-authorized + create-user API, RBAC, service defaults, and lifecycle semantics. diff --git a/deploy/buildkit/buildkitd.toml b/deploy/buildkit/buildkitd.toml index 0a3c3ce..6a3ebe6 100644 --- a/deploy/buildkit/buildkitd.toml +++ b/deploy/buildkit/buildkitd.toml @@ -21,7 +21,7 @@ [worker.oci] enabled = true rootless = true - no-process-sandbox = true + noProcessSandbox = false snapshotter = "overlayfs" # usa overlayfs se il kernel lo consente [worker.containerd] @@ -38,4 +38,4 @@ # Logging # ========================= [log] - level = "trace" \ No newline at end of file + level = "trace" diff --git a/deploy/user-provisioning/rbac.yaml b/deploy/user-provisioning/rbac.yaml new file mode 100644 index 0000000..51704b8 --- /dev/null +++ b/deploy/user-provisioning/rbac.yaml @@ -0,0 +1,65 @@ +# 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. +# + +# Optional feature permissions. Apply only when enabling user provisioning. +# Change the binding subject if Admin API uses a different service account. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: openserverless-user-provisioning + namespace: openserverless +rules: + - apiGroups: [""] + resources: [secrets] + resourceNames: [openserverless-user-provisioning-token] + verbs: [get] + - apiGroups: [openserverless.org] + resources: [whisks] + resourceNames: [controller] + verbs: [get] + - apiGroups: [openserverless.org] + resources: [whisksusers] + verbs: [get, create] + - apiGroups: [networking.k8s.io] + resources: [ingresses] + resourceNames: [apihost] + verbs: [get] + - apiGroups: [networking.k8s.io] + resources: [ingresses] + verbs: [create] + - apiGroups: [traefik.io] + resources: [middlewares] + verbs: [create] + - apiGroups: [""] + resources: [services] + resourceNames: [openserverless-streamer-api] + verbs: [get] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: openserverless-user-provisioning + namespace: openserverless +subjects: + - kind: ServiceAccount + name: openserverless-wsku + namespace: openserverless +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: openserverless-user-provisioning diff --git a/docs/OIDC.md b/docs/OIDC.md index 9a883d1..82c7b13 100644 --- a/docs/OIDC.md +++ b/docs/OIDC.md @@ -65,6 +65,28 @@ precedence: If a namespace cannot be provisioned automatically, an existing `WhiskUser` can be bound to an SSO identity manually by an admin, or auto-provisioning can be enabled (see below). +Namespace mapping does not grant ownership. Before returning credentials, admin-api requires +the `WhiskUser` annotations `openserverless.apache.org/sso-issuer` and +`openserverless.apache.org/sso-sub` to match the validated token's nonempty `iss` and `sub` +exactly. An unbound legacy namespace, a different subject, or a disabled binding receives +`403`, even when the external username matches the namespace. Password login is unaffected. + +To bind an existing namespace, a cluster administrator must verify its owner and obtain the +stable subject identifier from the trusted identity provider, then explicitly set: + +```sh +kubectl -n openserverless annotate whisksusers/ \ + openserverless.apache.org/sso-issuer='' \ + openserverless.apache.org/sso-sub='' --overwrite +``` + +This operation must not be delegated to an unauthenticated caller or derived from a matching +username alone. If the binding was deliberately disabled, re-enabling it requires a separate +administrator decision to remove or change `openserverless.apache.org/sso-disabled`. +Auto-provisioning records the binding on a new namespace and never takes over an unbound +existing namespace or one belonging to another subject. Binding is checked again after +provisioning. Orphan metadata and a `WhiskUser` replaced during the login are rejected. + ## Auto-provisioning By default, an OIDC login only succeeds if a `WhiskUser` for the resolved namespace already diff --git a/docs/USER_PROVISIONING.md b/docs/USER_PROVISIONING.md new file mode 100644 index 0000000..937fc8c --- /dev/null +++ b/docs/USER_PROVISIONING.md @@ -0,0 +1,211 @@ + +# User provisioning API + +The Admin API can create an OPS tenant on behalf of an external provisioning +system. It is independent +of tenant AUTH keys and SSO login. Existing login, CLI and SSO APIs retain their +behavior. Installing this image alone does not enable provisioning: the external +system must create the authorization Secret. + +## Enable and authorize + +The external deployment manages an `Opaque` Secret named +`openserverless-user-provisioning-token` in namespace `openserverless`. Its +`data.token` field contains the base64-encoded provisioning token. Generate at +least 32 random bytes of entropy. Do not use a tenant key, user password, or +service-account token. Only the external deployment and authorized operators +should have access to the value. + +For an isolated lab, an operator can create the Secret from a protected file: + +```sh +umask 077 +# Generate the token without a trailing newline; keep this file private. +python3 -c 'import secrets; print(secrets.token_hex(32), end="")' > provisioning-token +kubectl -n openserverless create secret generic openserverless-user-provisioning-token \ + --from-file=token=provisioning-token +kubectl apply -f deploy/user-provisioning/rbac.yaml +``` + +This is an explicit operator action, not application startup behavior. In a +managed installation let the external deployment own the Secret instead. +The optional Role/RoleBinding grants read access to this exact Secret, read +access to the controller configuration and base ingress, and create/read access +needed for users and their routes in `openserverless`. Check the service-account +subject against the actual Admin API pod. Existing broader roles are not changed +or narrowed by this manifest. Kubernetes RBAC cannot restrict `create` by +`resourceNames`; ingress and middleware creation is limited to this namespace. + +One background worker per Admin API process reads the Secret, then waits one +second before refreshing it again. HTTP requests never perform this Secret +lookup and never change its schedule. Each request compares its bearer in +constant time against the current, short-lived in-memory snapshot. Invalid +bearers return `401` and cannot start a shared cooldown or block valid bearers. +There is no per-caller token cache and no new persistent credential or service. + +Replacing or removing the Secret takes effect without a restart. A snapshot +expires two seconds after its Kubernetes read **started**, so rotation and +revocation have a maximum two-second stale-credential window. A slow response +cannot renew that window. An empty, malformed, missing or unreadable Secret +immediately clears the snapshot when observed. Expired snapshots are never used, +even while the next read is in progress. Requests already authorized may +complete. + +The first well-formed bearer starts the worker lazily. Until its first successful +read, or when no fresh credential is available, the endpoint returns `503` with +`Retry-After: 1`. Clients should retry with backoff. The failed request does not +cancel, delay or reset background refresh. Normal startup and other APIs do not +require the Secret. With multiple replicas, each process maintains its own +worker and freshness bound; ingress limits can enforce an overall deployment +traffic budget. + +## Request + +Use HTTPS outside isolated local testing. Send the token only through the +Authorization header, never in the URL. The caller should obtain the token from +its secret store and keep request bodies and headers out of application logs. + +```http +POST /system/api/v1/users +Authorization: Bearer +Content-Type: application/json + +{ + "user": "exampleuser", + "password": "", + "email": "user@example.org", + "services": "all" +} +``` + +`user`, `email`, and `password` must be non-empty strings without control +characters. Names follow the existing Admin API validator: 5–61 lowercase ASCII +letters/digits, without rewriting, with infrastructure names reserved. Email +must be an ASCII dot-atom `local@domain.tld` address (no quoted local parts or +backslash escapes), at most 254 characters, with a local part at most 64 +characters. This also avoids unsafe interpolation in the existing operator +metadata template. +When MongoDB is selected, user names are limited to 54 characters for FerretDB +role/database suffix compatibility; object storage limits them to 58 characters +so the generated bucket names fit the S3 limit. These checks occur before user +creation. Other selections retain the 61-character Admin API limit. + +Passwords are limited to 72 UTF-8 bytes to avoid bcrypt truncation; they are +never trimmed. Requests are limited to 16 KiB. Unknown properties are rejected. + +`services` is optional and defaults to `all`: all tenant services enabled in the +controller's `spec.components`. Missing optional components are treated as +disabled; the API never enables or installs cluster components. Explicit subsets +are comma-separated strings containing `redis`, `mongodb`, `postgres`, +`seaweedfs` and/or `milvus`. `minio` is accepted only on clusters whose controller +configuration explicitly enables that legacy backend. Both object-storage +backends cannot be enabled at the same time. + +SeaweedFS/MinIO selects the tenant data and website buckets. Existing quota +defaults are used, including storage quota `auto`. Unknown, disabled, duplicate, +empty, or mixed `all,redis` selections fail before creating anything. There is +no new quota override or "none" selector in this API. + +## Result and lifecycle + +```json +{ + "status": "ok", + "message": "User provisioning accepted", + "user": "exampleuser", + "provisioning_status": "pending", + "services": ["redis", "mongodb", "postgres", "seaweedfs", "milvus"] +} +``` + +HTTP `202` confirms that a new WhiskUser was created and its additional routes +were submitted. The operator asynchronously provisions the OpenWhisk subject, +service credentials, databases, buckets, profile metadata and standard tenant +API/static routes. The API creates CLI-equivalent additional Streamer and S3 +routes with WhiskUser owner references, so Kubernetes garbage-collects these +routes when the user is deleted. Streamer routes are added only if its service +exists. nginx and Traefik ingress classes are supported. S3 paths are never +rewritten, preserving SigV4 signatures. + +This is not a readiness guarantee. Use the existing administrator tooling to +inspect reconciliation and the existing password login/profile flow once it +finishes. No status endpoint or credentials are added to the create response. +The input password is retained as supplied, but all service secrets and tenant +AUTH values use cryptographically secure randomness. + +User creation uses POST/create, never apply or upsert. An existing, blocked, or +deleting WhiskUser returns `409`. Concurrent creates yield one winner and a +conflict; no existing account's password, services, or routes are changed. + +The API also verifies namespace availability directly in the existing CouchDB +databases before creating a WhiskUser. An existing subject, a deleted subject +tombstone, or retained OpenWhisk entity/activation IDs returns `409`. This +prevents a new credential set from inheriting a deleted tenant's actions, +packages or history. Even a tenant whose resources were removed normally keeps +its subject tombstone, so use a genuinely new name. The API never purges data or +tombstones automatically. Unavailable or malformed database responses return +`503` and create no user. This guard protects this provisioning endpoint; it +does not change the CLI/SSO lifecycle or certify their name-reuse behavior. + +If route submission fails after user creation, the API returns `503` with +`User created but route setup failed; administrator review required`. The user +is retained because reconciliation may already be running. Do not blindly +retry: the name now exists, so subsequent create calls return `409`. An +administrator should inspect the tenant, resolve ingress permissions/conflicts, +and reconcile the CLI routes (or explicitly remove this new tenant before +retrying). The API does not claim atomicity across multiple Kubernetes objects, +undo provisioning, overwrite existing routes, or perform a background retry. +A transport failure during user creation can also have an ambiguous outcome; +check the WhiskUser before attempting recovery. + +| HTTP | Meaning | +| --- | --- | +| 202 | New WhiskUser created; provisioning pending | +| 400 | Invalid payload or service selection | +| 401 | Missing or invalid provisioning Bearer token | +| 409 | User exists, name is retained, or concurrent creation conflict | +| 413 | Request exceeds 16 KiB | +| 415 | Request content type is not JSON | +| 503 | Feature unconfigured, dependency unavailable, or explicit partial route failure | +| 500 | Unexpected internal error | + +Errors use the existing `{"status":"ko","message":"..."}` envelope. +Authentication happens before account-existence checks. Passwords, provisioning +tokens, tenant AUTH values, and service credentials are neither logged nor +returned by the new path. See `/system/apidocs/` for the OpenAPI contract. + +## Validation + +Run `python -m unittest discover -s tests -v` in the project's Python environment. +The dedicated provisioning suite covers full/slim service resolution, subsets, +credential isolation, authorization, rotation/revocation, concurrent creation, +Kubernetes failure distinctions, malformed/oversized input, redacted logs, +partial route failures, and nginx/Traefik TLS/S3 route contracts. The legacy +client and login/SSO helpers are exercised by their existing suites. +It also covers deleted subject tombstones, orphaned entity/activation conflicts, +fail-closed availability checks, valid requests during repeated invalid bearers, +single-worker refresh, expiry during slow reads, and rotation/revocation recovery. + +Before publication, also verify the real Kubernetes RBAC and operator flow in +the lab: provisioning, readiness, password login, each enabled service, routes, +rotation, duplicate protection, and unchanged Alpha/Beta application suites. +MinIO and live nginx require separate installations; do not infer runtime +coverage for them from Traefik/SeaweedFS tests. diff --git a/openserverless/__init__.py b/openserverless/__init__.py index 7ae1020..9115f4e 100644 --- a/openserverless/__init__.py +++ b/openserverless/__init__.py @@ -44,6 +44,10 @@ "swagger_ui": True, "specs_route": "/system/apidocs/", "securityDefinitions": { + "provisioningBearer": { + "type": "apiKey", "name": "Authorization", "in": "header", + "description": "Bearer token from openserverless-user-provisioning-token" + }, "openwhiskBasicAuth": { "type": "apiKey", "name": "Authorization", @@ -80,4 +84,5 @@ def log_request_info(): import openserverless.rest.api import openserverless.rest.auth import openserverless.rest.build +import openserverless.rest.users diff --git a/openserverless/impl/auth/auth_service.py b/openserverless/impl/auth/auth_service.py index 9897a4f..fe95eea 100644 --- a/openserverless/impl/auth/auth_service.py +++ b/openserverless/impl/auth/auth_service.py @@ -127,6 +127,11 @@ def login_oidc(self, access_token, expected_namespace=None): logging.warning(f"OIDC token validation failed: {exc}") return res_builder.build_error_message("Invalid OIDC token", 401) + # A username is a routing hint, not a stable external identity. + if any(not isinstance(claims.get(key), str) or not claims[key].strip() + for key in ("iss", "sub")): + return res_builder.build_error_message("Invalid OIDC identity", 401) + username_claim = self._environ.get("OIDC_USERNAME_CLAIM", "preferred_username") external_username = claims[username_claim] try: @@ -145,12 +150,17 @@ def login_oidc(self, access_token, expected_namespace=None): 403, ) - if self.is_sso_login_disabled(login): - logging.warning(f"OIDC user {external_username} is unbound from namespace {login}") + whisk_user = self.kube_client.get_whisk_user(login) + if whisk_user and not self.is_sso_identity_bound(whisk_user, claims): + logging.warning(f"OIDC identity is not bound to namespace {login}") return res_builder.build_error_message("Forbidden", 403) user_data = self.fetch_user_data(login) + if user_data and not whisk_user: + # Orphan metadata must not be claimed through a concurrent create. + return res_builder.build_error_message("Forbidden", 403) + if not user_data: user_data = self.provision_oidc_user_if_enabled(login, external_username, claims) @@ -158,6 +168,17 @@ def login_oidc(self, access_token, expected_namespace=None): logging.warning(f"OIDC user {external_username} has no provisioned namespace") return res_builder.build_error_message("Namespace not provisioned", 404) + # Re-read after provisioning/waiting: an existing resource, including + # one found after a create conflict, must belong to this exact identity. + current_whisk_user = self.kube_client.get_whisk_user(login) + if not self.is_sso_identity_bound(current_whisk_user, claims): + return res_builder.build_error_message("Forbidden", 403) + if whisk_user and ( + (whisk_user.get("metadata") or {}).get("uid") + != (current_whisk_user.get("metadata") or {}).get("uid") + ): + return res_builder.build_error_message("Forbidden", 403) + login_data = self.map_login_data(user_data) if "AUTH" not in login_data: logging.error(f"OIDC user {login} metadata is missing AUTH") @@ -184,8 +205,7 @@ def provision_oidc_user_if_enabled(self, login, external_username, claims): 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 + return self.is_sso_identity_bound(existing_whisk_user, claims) whisk_user = self.build_sso_whisk_user(login, external_username, email, claims) if self.kube_client.create_whisk_user(whisk_user): @@ -193,9 +213,9 @@ def ensure_whisk_user(self, login, external_username, email, claims): 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 + existing_whisk_user = self.kube_client.get_whisk_user(login) + if existing_whisk_user: + return self.is_sso_identity_bound(existing_whisk_user, claims) logging.error(f"failed to create WhiskUser for OIDC user {login}") return False @@ -299,13 +319,18 @@ def _float_env(self, name, default): except (TypeError, ValueError): return default - def is_sso_login_disabled(self, login): - if not hasattr(self.kube_client, "get_whisk_user"): + def is_sso_identity_bound(self, whisk_user, claims): + if not 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)) + return ( + not self._is_truthy(annotations.get(SSO_DISABLED_ANNOTATION)) + and isinstance(claims.get("iss"), str) and bool(claims["iss"].strip()) + and isinstance(claims.get("sub"), str) and bool(claims["sub"].strip()) + and annotations.get(SSO_ISSUER_ANNOTATION) == claims["iss"] + and annotations.get(SSO_SUB_ANNOTATION) == claims["sub"] + ) def map_login_data(self, user_data): """ diff --git a/openserverless/impl/builder/build_service.py b/openserverless/impl/builder/build_service.py index 44dc85d..1990d77 100644 --- a/openserverless/impl/builder/build_service.py +++ b/openserverless/impl/builder/build_service.py @@ -25,6 +25,7 @@ import random import string import binascii +import re JOB_NAME = "build" CM_NAME = "cm" @@ -36,6 +37,19 @@ class BuildService: based on the provided build configuration. """ + @staticmethod + def valid_image_target(target, expected_user=None): + """Accept one OPS image name and a Docker-compatible tag.""" + if not isinstance(target, str): + return False + match = re.fullmatch(r"([a-z0-9](?:[a-z0-9._-]*[a-z0-9])?):([A-Za-z0-9_][A-Za-z0-9_.-]{0,127})", target) + return bool(match and (expected_user is None or match.group(1) == expected_user)) + + @staticmethod + def valid_source_image(source): + """Keep FROM limited to a single image reference, never Dockerfile text.""" + return isinstance(source, str) and bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/@-]*", source)) + def __init__(self, user_env=None): # A super userful Kube Api Client self.kube_client = KubeApiClient() @@ -206,6 +220,11 @@ def build(self, image_name: str) -> tuple[bool, str]: import tempfile import base64 + if not self.valid_image_target(image_name): + return (False, "Invalid target for the build") + if not self.valid_source_image(self.build_config.get("source")): + return (False, "Invalid source image for the build") + # define registry host self.registry_host = self.get_registry_host() if self.registry_host is None: @@ -402,11 +421,16 @@ def check_build_dir(self, unzip_dir: str) -> bool: def create_build_job(self, image_name: str) -> dict: """Create a Kubernetes job manifest for building the Docker image.""" + if not self.valid_image_target(image_name): + raise ValueError("Invalid target for the build") if not self.custom_registry_auth: registry_image_name = f"{self.registry_host}/{image_name}" else: registry_image_name = f"{image_name}" + # Require a private Kubernetes user namespace; keep BuildKit's process + # sandbox enabled. Subordinate IDs exclude Pod root and the worker UID. + # The resulting nested map supports build IDs 0..65534. # --- MANIFEST DEL JOB --- job_manifest = { "apiVersion": "batch/v1", @@ -417,6 +441,8 @@ def create_build_job(self, image_name: str) -> dict: "template": { "spec": { "restartPolicy": "Never", + "automountServiceAccountToken": False, + "hostUsers": False, "volumes": [ { "name": "openserverless-buildkitd-conf", @@ -449,7 +475,7 @@ def create_build_job(self, image_name: str) -> dict: "name": "copy-build-context", "image": "busybox:1.36", "command": [ - "sh", "-c", "cp -rvL /configmap/* /workspace/ && cat /workspace/Dockerfile", + "sh", "-c", "cp -rL /configmap/* /workspace/ && printf 'user:1:999\\nuser:1001:64535\\n' > /workspace/subuid && cp /workspace/subuid /workspace/subgid", ], "volumeMounts": [ { "name": "build-context-vol", "mountPath": "/configmap", }, @@ -460,25 +486,30 @@ def create_build_job(self, image_name: str) -> dict: "containers": [ { "name": "buildkit", - "image": "moby/buildkit:master-rootless", + "image": "moby/buildkit@sha256:e6102bb86f8a2c79c8ae8aa0de3d5edef500dd33b619414413e06138e13fb1a1", "command": ["sh", "-c"], "args": [ "rootlesskit buildkitd --config /config/buildkitd.toml & sleep 3 && " - f"buildctl build --progress=plain --frontend=dockerfile.v0 --local context=/workspace --local dockerfile=/workspace --output=type=image,name={registry_image_name},push=true" + 'buildctl build --progress=plain --frontend=dockerfile.v0 --local context=/workspace --local dockerfile=/workspace --output="type=image,name=${REGISTRY_IMAGE_NAME},push=true"' ], "securityContext": { "runAsUser": 1000, "runAsGroup": 1000, "allowPrivilegeEscalation": True, - "privileged": True, - "seccompProfile": { "type": "Unconfined" } + "privileged": False, + "procMount": "Unmasked", + "seccompProfile": { "type": "Unconfined" }, + "appArmorProfile": { "type": "Unconfined" }, }, "env": [ - { "name": "BUILDKIT_ROOTLESS", "value": "1" } + { "name": "BUILDKIT_ROOTLESS", "value": "1" }, + { "name": "REGISTRY_IMAGE_NAME", "value": registry_image_name } ], "volumeMounts": [ { "name": "openserverless-buildkitd-conf", "mountPath": "/config" }, { "name": "workspace", "mountPath": "/workspace" }, + { "name": "workspace", "mountPath": "/etc/subuid", "subPath": "subuid", "readOnly": True }, + { "name": "workspace", "mountPath": "/etc/subgid", "subPath": "subgid", "readOnly": True }, { "name": "docker-config", "mountPath": "/home/user/.docker" }, { "name": "img-cache", "mountPath": "/tmp" }, { "name": "cdi-etc", "mountPath": "/etc/cdi" }, diff --git a/openserverless/impl/provisioning/__init__.py b/openserverless/impl/provisioning/__init__.py new file mode 100644 index 0000000..d0a78df --- /dev/null +++ b/openserverless/impl/provisioning/__init__.py @@ -0,0 +1,17 @@ +# 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. +# diff --git a/openserverless/impl/provisioning/admission.py b/openserverless/impl/provisioning/admission.py new file mode 100644 index 0000000..4ef9b87 --- /dev/null +++ b/openserverless/impl/provisioning/admission.py @@ -0,0 +1,102 @@ +# 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. +# + +"""Request-independent refresh of a short-lived provisioning credential.""" +import base64 +import binascii +import secrets +import threading +import time + +from .kube_client import ProvisioningError, ProvisioningKubeClient + + +def decode_provisioning_token(secret): + try: + if not isinstance(secret, dict) or secret.get("type") != "Opaque": + raise ValueError() + token = base64.b64decode(secret["data"]["token"], validate=True) + if not token.strip() or len(token) > 8192: + raise ValueError() + return token + except (KeyError, TypeError, ValueError, binascii.Error): + raise ProvisioningError("User provisioning authorization is not configured", 503) from None + + +def _load_secret(): + return ProvisioningKubeClient().provisioning_secret() + + +class ProvisioningAuthorizer: + REFRESH_SECONDS = 1.0 + MAX_AGE_SECONDS = 2.0 + + def __init__(self, loader=_load_secret, clock=time.monotonic): + self._loader = loader + self._clock = clock + self._lock = threading.Lock() + self._snapshot = None + self._thread = None + self._stop = threading.Event() + + def start(self): + # Starting is lazy so importing the app and using unrelated routes do + # not require Kubernetes or the optional provisioning Secret. + with self._lock: + if self._thread is None and not self._stop.is_set(): + worker = threading.Thread(target=self._run, name="provisioning-credential", daemon=True) + worker.start() + self._thread = worker + + def close(self): + self._stop.set() + with self._lock: + self._snapshot = None + + def _refresh_once(self): + # Age starts before I/O: a delayed response must not extend the lifetime + # of credentials read before rotation or revocation. + started = self._clock() + try: + token = decode_provisioning_token(self._loader()) + expires = started + self.MAX_AGE_SECONDS + snapshot = (token, expires) if self._clock() < expires else None + except Exception: + # Dependency/configuration failures revoke the local snapshot. + # Do not retain exception bodies, backend URLs, or credential data. + snapshot = None + with self._lock: + self._snapshot = None if self._stop.is_set() else snapshot + + def _run(self): + while not self._stop.is_set(): + self._refresh_once() + if self._stop.wait(self.REFRESH_SECONDS): + return + + def authorize(self, supplied): + self.start() + with self._lock: + snapshot = self._snapshot + if snapshot is None or self._clock() >= snapshot[1]: + raise ProvisioningError("Provisioning authorization temporarily unavailable", 503) + if not secrets.compare_digest(supplied, snapshot[0]): + raise ProvisioningError("Missing or invalid provisioning token", 401) + + +provisioning_authorizer = ProvisioningAuthorizer() diff --git a/openserverless/impl/provisioning/kube_client.py b/openserverless/impl/provisioning/kube_client.py new file mode 100644 index 0000000..8eee3d1 --- /dev/null +++ b/openserverless/impl/provisioning/kube_client.py @@ -0,0 +1,81 @@ +# 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. +# + +"""Bounded Kubernetes operations for provisioning; never log response bodies.""" +import requests + +from openserverless.common.kube_api_client import KubeApiClient + + +class ProvisioningError(Exception): + def __init__(self, message, status=503): + super().__init__(message) + self.status = status + + +class ProvisioningKubeClient(KubeApiClient): + NAMESPACE = "openserverless" + USERS = "/apis/openserverless.org/v1/namespaces/openserverless/whisksusers" + INGRESS = "/apis/networking.k8s.io/v1/namespaces/openserverless/ingresses" + MIDDLEWARE = "/apis/traefik.io/v1alpha1/namespaces/openserverless/middlewares" + CORE = "/api/v1/namespaces/openserverless" + + def _request(self, method, path, body=None, missing_ok=False): + try: + response = requests.request( + method, self.host + path, + headers={"Authorization": self.token}, json=body, + verify=self.ssl_ca_cert, timeout=(5, 20), + ) + except requests.RequestException: + raise ProvisioningError("Provisioning dependency unavailable") from None + if response.status_code == 404 and missing_ok: + return None + if method == "POST" and response.status_code == 409: + raise ProvisioningError("Resource already exists", 409) + if response.status_code not in (200, 201, 202): + raise ProvisioningError("Provisioning dependency unavailable") + try: + result = response.json() + except ValueError: + raise ProvisioningError("Invalid provisioning dependency response") from None + if not isinstance(result, dict): + raise ProvisioningError("Invalid provisioning dependency response") + return result + + def provisioning_secret(self): + return self._request("GET", self.CORE + "/secrets/openserverless-user-provisioning-token", missing_ok=True) + + def cluster(self): + return self._request("GET", "/apis/openserverless.org/v1/namespaces/openserverless/whisks/controller") + + def apihost_ingress(self): + return self._request("GET", self.INGRESS + "/apihost") + + def streamer_service(self): + return self._request("GET", self.CORE + "/services/openserverless-streamer-api", missing_ok=True) + + def existing_user(self, name): + return self._request("GET", self.USERS + "/" + name, missing_ok=True) + + def create_user(self, body): + return self._request("POST", self.USERS, body) + + def create_route(self, body): + path = self.MIDDLEWARE if body["kind"] == "Middleware" else self.INGRESS + return self._request("POST", path, body) diff --git a/openserverless/impl/provisioning/namespace_guard.py b/openserverless/impl/provisioning/namespace_guard.py new file mode 100644 index 0000000..f31e5a8 --- /dev/null +++ b/openserverless/impl/provisioning/namespace_guard.py @@ -0,0 +1,70 @@ +# 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. +# + +"""Reject OpenWhisk namespace reuse, including deleted subject tombstones.""" +import json + +import requests + +from openserverless.couchdb.couchdb_util import CouchDB +from .kube_client import ProvisioningError + + +class NamespaceGuard: + def assert_unused(self, name): + # Called only after authorization and strict OPS name validation. Use + # existing CouchDB configuration; no new credentials or database. + try: + db = CouchDB() + except ValueError: + raise ProvisioningError("Namespace availability could not be verified") from None + try: + # Unlike GET /subjects/name (404 for a tombstone), _all_docs with + # keys retains the deleted row. Never release a used identity just + # because its WhiskUser or live subject document disappeared. + rows = self._rows(db, "subjects", {"keys": json.dumps([name])}) + if len(rows) != 1 or not isinstance(rows[0], dict) or rows[0].get("key") != name: + raise ProvisioningError("Namespace availability could not be verified") + if rows[0].get("error") != "not_found": + raise ProvisioningError("User name is already used or retained; administrator review required", 409) + # Also cover imported/orphaned entities whose subject was purged. + # OpenWhisk entity and activation IDs are namespace/name. The + # exclusive ASCII upper bound covers nested packages as well. + params = {"startkey": json.dumps(name + "/"), + "endkey": json.dumps(name + "0"), + "inclusive_end": "false", "limit": 1} + for database in ("whisks", "activations"): + if self._rows(db, database, params): + raise ProvisioningError("User name is already used or retained; administrator review required", 409) + finally: + db.db_session.close() + + @staticmethod + def _rows(db, database, params): + try: + response = db.db_session.get(f"{db.db_base}{database}/_all_docs", + params=params, timeout=(2, 3)) + if response.status_code != 200: + raise ValueError() + body = response.json() + if not isinstance(body, dict) or not isinstance(body.get("rows"), list): + raise ValueError() + return body["rows"] + except (requests.RequestException, ValueError): + # Do not expose backend bodies, credentials or diagnostic URLs. + raise ProvisioningError("Namespace availability could not be verified") from None diff --git a/openserverless/impl/provisioning/routes.py b/openserverless/impl/provisioning/routes.py new file mode 100644 index 0000000..ce6f774 --- /dev/null +++ b/openserverless/impl/provisioning/routes.py @@ -0,0 +1,97 @@ +# 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. +# + +"""Tenant routes matching the CLI's nginx/Traefik streaming and S3 paths.""" +import re +from copy import deepcopy + +from .kube_client import ProvisioningError + + +def route_context(apihost): + spec = apihost.get("spec", {}) + ingress_class = spec.get("ingressClassName") + if ingress_class is None: + ingress_class = apihost.get("metadata", {}).get("annotations", {}).get("kubernetes.io/ingress.class") + rules = spec.get("rules", []) + hostname = rules[0].get("host", "") if rules else "" + if ingress_class not in ("nginx", "traefik") or not isinstance(hostname, str) or not re.fullmatch( + r"[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", hostname + ): + raise ProvisioningError("Unsupported or unavailable cluster ingress configuration") + return ingress_class, hostname, bool(spec.get("tls")) + + +def tenant_routes(user, context, services, streamer, uid): + ingress_class, domain, tls = context + hostname = f"{user}.{domain}" + if len(hostname) > 253: + raise ProvisioningError("Tenant hostname is too long", 400) + owner = [{"apiVersion": "openserverless.org/v1", "kind": "WhiskUser", + "name": user, "uid": uid}] + annotations = {"nginx.ingress.kubernetes.io/proxy-body-size": "48m"} + if ingress_class == "nginx": + annotations.update({f"nginx.ingress.kubernetes.io/proxy-{key}-timeout": "300" + for key in ("connect", "read", "send")}) + else: + annotations.update({f"traefik.ingress.kubernetes.io/transport.respondingTimeouts.{key}Timeout": "600" + for key in ("idle", "read", "write")}) + if tls: + annotations["cert-manager.io/cluster-issuer"] = "letsencrypt-prod" + if ingress_class == "nginx": + annotations["nginx.ingress.kubernetes.io/force-ssl-redirect"] = "true" + else: + annotations["traefik.ingress.kubernetes.io/router.tls"] = "true" + annotations["traefik.ingress.kubernetes.io/router.entrypoints"] = "websecure,web" + + def ingress(suffix, paths, extra=None): + meta = {"name": f"{user}-{suffix}", "namespace": "openserverless", + "ownerReferences": deepcopy(owner), "annotations": {**annotations, **(extra or {})}} + spec = {"ingressClassName": ingress_class, + "rules": [{"host": hostname, "http": {"paths": paths}}]} + if tls: + spec["tls"] = [{"hosts": [hostname], "secretName": "streamer-api-tls"}] + return {"apiVersion": "networking.k8s.io/v1", "kind": "Ingress", "metadata": meta, "spec": spec} + + def path(value, service="openserverless-streamer-api", port=8080, path_type="Prefix"): + return {"path": value, "pathType": path_type, + "backend": {"service": {"name": service, "port": {"number": port}}}} + + result = [] + if streamer: + result.append(ingress("my-streamer-ingress", [path("/web"), path("/action")])) + for mode in ("web", "action"): + middleware_name = f"{user}-stream-{mode}-rewrite" + if ingress_class == "traefik": + result.append({"apiVersion": "traefik.io/v1alpha1", "kind": "Middleware", + "metadata": {"name": middleware_name, "namespace": "openserverless", "ownerReferences": deepcopy(owner)}, + "spec": {"replacePathRegex": {"regex": f"/stream/{mode}(/|$)(.*)", + "replacement": f"/{mode}/{user}/$2"}}}) + extra = {"traefik.ingress.kubernetes.io/router.middlewares": f"openserverless-{middleware_name}@kubernetescrd"} + paths = [path(f"/stream/{mode}")] + else: + extra = {"nginx.ingress.kubernetes.io/use-regex": "true", + "nginx.ingress.kubernetes.io/rewrite-target": f"/{mode}/{user}/$2"} + paths = [path(f"/stream/{mode}(/|$)(.*)", path_type="ImplementationSpecific")] + result.append(ingress(f"my-streamer-{mode}-ingress", paths, extra)) + if "seaweedfs" in services or "minio" in services: + backend = "seaweedfs" if "seaweedfs" in services else "minio" + result.append(ingress("my-s3-ingress", [path(f"/{user}-web", backend, 9000), + path(f"/{user}-data", backend, 9000)], + {"nginx.ingress.kubernetes.io/proxy-body-size": "1024m"})) + return result diff --git a/openserverless/impl/provisioning/user_service.py b/openserverless/impl/provisioning/user_service.py new file mode 100644 index 0000000..52cf389 --- /dev/null +++ b/openserverless/impl/provisioning/user_service.py @@ -0,0 +1,146 @@ +# 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. +# + +"""Externally authorized, create-only OPS tenant provisioning.""" +import re +import secrets +import string +import uuid + +from openserverless.common.validation import is_valid_username +from .kube_client import ProvisioningError +from .namespace_guard import NamespaceGuard +from .routes import route_context, tenant_routes + +SECRET_NAME = "openserverless-user-provisioning-token" +SERVICE_NAMES = ("redis", "mongodb", "postgres", "seaweedfs", "minio", "milvus") +# Infrastructure identities must not be provisioned as ordinary tenants. +RESERVED_USERS = {"openserverless", "whisk", "system", "guest", "default", "controller", "invoker"} + + +def bearer_token(header): + if not isinstance(header, str) or len(header) > 8192: + raise ProvisioningError("Missing or invalid provisioning token", 401) + parts = header.split() + if len(parts) != 2 or parts[0].lower() != "bearer" or not parts[1].isascii(): + raise ProvisioningError("Missing or invalid provisioning token", 401) + return parts[1].encode("ascii") + + +def random_secret(length=32): + return "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(length)) + + +class UserProvisioningService: + def __init__(self, kube): + self.kube = kube + + def validate(self, body): + if not isinstance(body, dict) or set(body) - {"user", "password", "email", "services"}: + raise ProvisioningError("Invalid user provisioning payload", 400) + for field in ("user", "password", "email"): + value = body.get(field) + if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 or ord(c) == 127 for c in value): + raise ProvisioningError(f"Invalid or missing {field}", 400) + user, email = body["user"], body["email"] + if not is_valid_username(user) or user in RESERVED_USERS: + raise ProvisioningError("Invalid or reserved user name", 400) + # The operator embeds email in its JSON metadata template. Restrict to + # an ordinary ASCII dot-atom address (no quoted local part or escapes). + local, separator, domain = email.rpartition("@") + local_atom = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+" + domain_label = r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?" + if (len(email) > 254 or not separator or len(local) > 64 + or not re.fullmatch(local_atom + r"(?:\." + local_atom + r")*", local) + or not re.fullmatch(domain_label + r"(?:\." + domain_label + r")+", domain)): + raise ProvisioningError("Invalid email", 400) + # CouchDB uses bcrypt; do not silently truncate UTF-8 passwords. + try: + encoded_password = body["password"].encode("utf-8") + email.encode("utf-8") + except UnicodeError: + raise ProvisioningError("Invalid UTF-8 field", 400) from None + if len(encoded_password) > 72: + raise ProvisioningError("Password exceeds 72 UTF-8 bytes", 400) + selection = body.get("services", "all") + if not isinstance(selection, str): + raise ProvisioningError("Invalid services selection", 400) + selected = [item.strip() for item in selection.split(",")] + if (not all(selected) or len(set(selected)) != len(selected) + or ("all" in selected and len(selected) != 1) + or any(item not in SERVICE_NAMES + ("all",) for item in selected)): + raise ProvisioningError("Invalid services selection", 400) + return selected + + def provision(self, body): + selection = self.validate(body) + user = body["user"] + if self.kube.existing_user(user) is not None: + raise ProvisioningError("User already exists", 409) + cluster = self.kube.cluster() + components = cluster.get("spec", {}).get("components") + if not isinstance(components, dict) or any( + not isinstance(components.get(name, False), bool) for name in SERVICE_NAMES + ): + raise ProvisioningError("Cluster service configuration unavailable") + available = [name for name in SERVICE_NAMES if components.get(name) is True] + if "seaweedfs" in available and "minio" in available: + raise ProvisioningError("Ambiguous object-storage configuration") + services = available if selection == ["all"] else [name for name in SERVICE_NAMES if name in selection] + if any(name not in available for name in services): + raise ProvisioningError("Requested service is not enabled in this cluster", 400) + # Match the backend identifier limits before submitting an unusable CR. + # FerretDB appends "_ferretdb" to PostgreSQL role/database identifiers. + if "mongodb" in services and len(user) > 54: + raise ProvisioningError("MongoDB users must not exceed 54 characters", 400) + if any(name in services for name in ("seaweedfs", "minio")) and len(user) > 58: + raise ProvisioningError("Object-storage users must not exceed 58 characters", 400) + context = route_context(self.kube.apihost_ingress()) + streamer = self.kube.streamer_service() is not None + # Validate hostname and route construction before creating anything. + tenant_routes(user, context, services, streamer, "preflight") + NamespaceGuard().assert_unused(user) + spec = {"namespace": user, "email": body["email"], "password": body["password"], + "auth": f"{uuid.uuid4()}:{random_secret(64)}"} + for name in ("redis", "mongodb", "postgres", "milvus"): + if name in services: + spec[name] = {"enabled": True, "password": random_secret(), + "prefix" if name == "redis" else "database": user} + if "seaweedfs" in services or "minio" in services: + spec["object-storage"] = {"password": random_secret(40), "quota": "auto", + "data": {"enabled": True, "bucket": f"{user}-data"}, + "route": {"enabled": True, "bucket": f"{user}-web"}} + resource = {"apiVersion": "openserverless.org/v1", "kind": "WhiskUser", + "metadata": {"name": user, "namespace": "openserverless"}, "spec": spec} + try: + created = self.kube.create_user(resource) + except ProvisioningError as exc: + if exc.status == 409: + raise ProvisioningError("User already exists", 409) from None + raise + uid = created.get("metadata", {}).get("uid") + if not isinstance(uid, str) or not uid: + raise ProvisioningError("User creation could not be confirmed; administrator review required") + try: + for route in tenant_routes(user, context, services, streamer, uid): + self.kube.create_route(route) + except ProvisioningError: + # Do not roll back a tenant which the operator may already be using. + raise ProvisioningError("User created but route setup failed; administrator review required") from None + return {"status": "ok", "message": "User provisioning accepted", "user": user, + "provisioning_status": "pending", "services": services} diff --git a/openserverless/rest/build.py b/openserverless/rest/build.py index 066ab60..c74ed37 100644 --- a/openserverless/rest/build.py +++ b/openserverless/rest/build.py @@ -145,12 +145,12 @@ def build(): # validate the target wsk_user_name = auth_result.get('login','').lower() target = json_data.get('target') - target_user = str(target).split(':')[0] - # Strict user check is enabled by default for security strict_user_check = os.environ.get("STRICT_USER_CHECK", "true").lower() not in ("false", "0", "no", "off") - if strict_user_check and (wsk_user_name != target_user): + if not BuildService.valid_image_target(target, wsk_user_name if strict_user_check else None): return res_builder.build_error_message("Invalid target for the build.", status_code=HTTPStatus.BAD_REQUEST) + if not BuildService.valid_source_image(json_data.get('source')): + return res_builder.build_error_message("Invalid source image for the build.", status_code=HTTPStatus.BAD_REQUEST) env['wsk_user_name'] = wsk_user_name build_service = BuildService(user_env=env) @@ -251,4 +251,4 @@ def clean(): if clean_result == -1: return res_builder.build_error_message("Failed to clean up old build jobs.", status_code=HTTPStatus.INTERNAL_SERVER_ERROR) - return res_builder.build_response_message(f"Cleaned up {clean_result} jobs successfully.", status_code=HTTPStatus.OK) \ No newline at end of file + return res_builder.build_response_message(f"Cleaned up {clean_result} jobs successfully.", status_code=HTTPStatus.OK) diff --git a/openserverless/rest/users.py b/openserverless/rest/users.py new file mode 100644 index 0000000..6534361 --- /dev/null +++ b/openserverless/rest/users.py @@ -0,0 +1,95 @@ +# 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. +# + +"""Administrative user provisioning with a separately managed capability.""" +import logging + +from flask import request +from werkzeug.exceptions import BadRequest, RequestEntityTooLarge + +from openserverless import app +from openserverless.common import response_builder +from openserverless.error.config_exception import ConfigException +from openserverless.impl.provisioning.admission import provisioning_authorizer +from openserverless.impl.provisioning.kube_client import ProvisioningError, ProvisioningKubeClient +from openserverless.impl.provisioning.user_service import UserProvisioningService, bearer_token + + +@app.route("/system/api/v1/users", methods=["POST"]) +def create_user(): + """ + Create an OPS tenant (asynchronous provisioning). + --- + tags: [User provisioning] + operationId: createUser + security: + - provisioningBearer: [] + consumes: [application/json] + parameters: + - in: header + name: Authorization + required: true + type: string + description: Bearer token from the externally managed provisioning Secret. + - in: body + name: user + required: true + schema: + type: object + additionalProperties: false + required: [user, password, email] + properties: + user: {type: string, minLength: 5, maxLength: 61} + password: {type: string, format: password} + email: {type: string, format: email} + services: {type: string, default: all} + responses: + 202: {description: "WhiskUser created; provisioning pending, not a readiness guarantee."} + 400: {description: Invalid payload or unavailable requested service.} + 401: {description: Missing or invalid provisioning token.} + 409: {description: User exists or namespace is retained; no update performed.} + 413: {description: Request body exceeds 16 KiB.} + 415: {description: Content-Type must be application/json.} + 503: {description: "Provisioning configuration or Kubernetes dependency unavailable."} + 500: {description: Unexpected internal error.} + """ + try: + token = bearer_token(request.headers.get("Authorization")) + provisioning_authorizer.authorize(token) + service = UserProvisioningService(ProvisioningKubeClient()) + if not request.is_json: + raise ProvisioningError("Content-Type must be application/json", 415) + request.max_content_length = 16384 + body = request.get_json() + result = service.provision(body) + return response_builder.build_response_with_data(result, 202) + except ProvisioningError as exc: + response = response_builder.build_error_message(str(exc), exc.status) + if exc.status == 503: + response.headers["Retry-After"] = "1" + return response + except BadRequest: + return response_builder.build_error_message("Invalid JSON payload", 400) + except RequestEntityTooLarge: + return response_builder.build_error_message("Request body exceeds 16 KiB", 413) + except (ConfigException, OSError): + return response_builder.build_error_message("Provisioning dependency unavailable", 503) + except Exception as exc: + # Exception messages/tracebacks may contain caller data; log only type. + logging.error("User provisioning failed (%s)", type(exc).__name__) + return response_builder.build_error_message("User provisioning failed", 500) diff --git a/tests/test_auth_service_oidc.py b/tests/test_auth_service_oidc.py index b139f26..808e0f3 100644 --- a/tests/test_auth_service_oidc.py +++ b/tests/test_auth_service_oidc.py @@ -55,6 +55,8 @@ def get_whisk_user(self, username): def create_whisk_user(self, whisk_user): self.created.append(whisk_user) + if self.create_result: + self.existing = whisk_user return self.create_result @@ -69,12 +71,22 @@ def service(self, docs, environ=None, kube_client=None): return AuthService( environ=env, couch_db=FakeCouchDB(docs), - kube_client=kube_client if kube_client is not None else object(), + kube_client=kube_client if kube_client is not None else FakeKubeClient( + existing={ + "metadata": {"annotations": { + "openserverless.apache.org/sso-issuer": "http://issuer.test", + "openserverless.apache.org/sso-sub": "owner-subject", + }}, + "spec": {"auth": "uuid:key"}, + } + ), ) @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 = { + "iss": "http://issuer.test", + "sub": "owner-subject", "preferred_username": "devel", } service = self.service( @@ -109,6 +121,8 @@ def test_oidc_login_returns_ops_login_data(self, validator_class): @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 = { + "iss": "http://issuer.test", + "sub": "owner-subject", "preferred_username": "missing.lab", } service = self.service([]) @@ -122,6 +136,8 @@ def test_oidc_login_returns_404_when_namespace_is_missing(self, validator_class) @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 = { + "iss": "http://issuer.test", + "sub": "owner-subject", "preferred_username": "devel", } service = self.service( @@ -183,7 +199,7 @@ def test_oidc_login_autoprovisions_whisk_user_when_enabled(self, validator_class kube_client=kube_client, ) - with app.app_context(): + with app.app_context(), patch.object(service, "_random_auth", return_value="uuid:key"): response = service.login_oidc("token") self.assertEqual(200, response.status_code) @@ -268,7 +284,7 @@ def test_oidc_login_autoprovisions_normalized_namespace(self, validator_class): kube_client=kube_client, ) - with app.app_context(): + with app.app_context(), patch.object(service, "_random_auth", return_value="uuid:key"): response = service.login_oidc("token") self.assertEqual(200, response.status_code) @@ -286,6 +302,8 @@ def test_oidc_login_autoprovisions_normalized_namespace(self, validator_class): @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 = { + "iss": "http://issuer.test", + "sub": "owner-subject", "preferred_username": "developer.lab", } service = self.service( diff --git a/tests/test_build_api_security.py b/tests/test_build_api_security.py new file mode 100644 index 0000000..d546e98 --- /dev/null +++ b/tests/test_build_api_security.py @@ -0,0 +1,52 @@ +# 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, Mock +from openserverless import app +from openserverless.rest import build +from openserverless.impl.builder.build_service import BuildService + +class BuildApiSecurityTest(unittest.TestCase): + def setUp(self): + self.client = app.test_client() + self.payload = {"source":"runtime:python", "target":"alpha:release", "kind":"python"} + + @patch.object(build, "env_to_dict", return_value={"SYNTHETIC":"value"}) + @patch.object(build, "authorize", return_value={"login":"alpha"}) + def test_invalid_inputs_rejected_before_worker_creation(self, authorize, env): + with patch.dict(build.os.environ, {"STRICT_USER_CHECK":"true"}), patch.object(BuildService, "__init__", return_value=None) as initialize: + for key, value in (("target","alpha:ok;id"), ("target","beta:release"), + ("target","alpha:ok$(id)"), ("source","runtime:python\nRUN id")): + response = self.client.post("/system/api/v1/build/start",json={**self.payload,key:value}) + self.assertEqual(400,response.status_code) + initialize.assert_not_called() + + @patch.object(build, "env_to_dict", return_value={"SYNTHETIC":"value"}) + @patch.object(build, "authorize", return_value={"login":"alpha"}) + def test_valid_request_keeps_response_contract(self, authorize, env): + service = Mock(id="synthetic-id", job_name="build-alpha-test") + service.build.return_value = (True,"build-alpha-test") + with patch.object(build,"BuildService") as factory: + factory.valid_image_target.side_effect = BuildService.valid_image_target + factory.valid_source_image.side_effect = BuildService.valid_source_image + factory.return_value = service + response = self.client.post("/system/api/v1/build/start",json=self.payload) + self.assertEqual(200,response.status_code) + self.assertEqual("build-alpha-test",response.json["job_name"]) + service.init.assert_called_once_with(build_config=self.payload) + service.build.assert_called_once_with("alpha:release") diff --git a/tests/test_build_target_security.py b/tests/test_build_target_security.py new file mode 100644 index 0000000..152d583 --- /dev/null +++ b/tests/test_build_target_security.py @@ -0,0 +1,76 @@ +# 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 Mock + +from openserverless.impl.builder.build_service import BuildService + +class BuildTargetSecurityTest(unittest.TestCase): + def test_rejects_shell_syntax_before_kubernetes_calls(self): + service = BuildService.__new__(BuildService) + service.user = "alpha" + service.kube_client = Mock() + for target in ("alpha:ok;id", "alpha:ok$(id)", "alpha:ok\nRUN id", "alpha:ok,foo=bar"): + self.assertEqual((False, "Invalid target for the build"), service.build(target)) + service.kube_client.assert_not_called() + self.assertEqual([], service.kube_client.mock_calls) + + def test_valid_target_is_passed_as_data_to_buildctl(self): + service = BuildService.__new__(BuildService) + service.user = "alpha" + service.registry_host = "registry.local:5000" + service.custom_registry_auth = False + service.job_name = "build-alpha-test" + service.cm = "cm-alpha-test" + service.registry_auth = "registry-pull-secret" + job = service.create_build_job("alpha:release_1.2") + buildkit = job["spec"]["template"]["spec"]["containers"][0] + self.assertEqual("registry.local:5000/alpha:release_1.2", next(x["value"] for x in buildkit["env"] if x["name"] == "REGISTRY_IMAGE_NAME")) + self.assertIn('${REGISTRY_IMAGE_NAME}', buildkit["args"][0]) + self.assertNotIn("alpha:release_1.2", buildkit["args"][0]) + pod = job["spec"]["template"]["spec"] + self.assertFalse(pod["automountServiceAccountToken"]) + self.assertFalse(pod["hostUsers"]) + self.assertEqual("Unmasked", buildkit["securityContext"]["procMount"]) + self.assertIn("@sha256:", buildkit["image"]) + self.assertEqual({"/etc/subuid", "/etc/subgid"}, { + mount["mountPath"] for mount in buildkit["volumeMounts"] if mount.get("subPath") + }) + self.assertFalse(buildkit["securityContext"]["privileged"]) + self.assertEqual("Unconfined", buildkit["securityContext"]["appArmorProfile"]["type"]) + self.assertEqual("Unconfined", buildkit["securityContext"]["seccompProfile"]["type"]) + + def test_source_cannot_add_dockerfile_instructions(self): + self.assertTrue(BuildService.valid_source_image("docker.io/apache/runtime:1.2")) + self.assertFalse(BuildService.valid_source_image("docker.io/apache/runtime:1.2\nRUN id")) + self.assertFalse(BuildService.valid_image_target("beta:release", "alpha")) + + def test_dependency_build_keeps_existing_extension_contract(self): + service = BuildService.__new__(BuildService) + service.build_config = {"source": "runtime:python", "kind": "python", "file": "encoded"} + dockerfile = service.create_docker_file("requirements.txt") + self.assertIn("RUN /bin/extend\n", dockerfile) + self.assertTrue(dockerfile.endswith("USER nobody\n")) + + def test_base_image_only_build_does_not_install_dependencies(self): + service = BuildService.__new__(BuildService) + service.build_config = {"source": "runtime:python", "kind": "python"} + self.assertEqual("FROM runtime:python\n\n", service.create_docker_file()) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sso_identity_binding.py b/tests/test_sso_identity_binding.py new file mode 100644 index 0000000..9b6e879 --- /dev/null +++ b/tests/test_sso_identity_binding.py @@ -0,0 +1,141 @@ +# 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 copy +import unittest +from unittest.mock import Mock, patch + +from openserverless import app +from openserverless.impl.auth.auth_service import AuthService + + +class SsoIdentityBindingTest(unittest.TestCase): + def setUp(self): + self.claims = {"iss": "https://issuer.test/realm", "sub": "owner-id", + "preferred_username": "tenantuser", "email": "user@example.test"} + self.resource = {"metadata": {"name": "tenantuser", "uid": "tenant-uid", "annotations": { + "openserverless.apache.org/sso-issuer": self.claims["iss"], + "openserverless.apache.org/sso-sub": self.claims["sub"], + }}, "spec": {"namespace": "tenantuser", "auth": "uuid:key"}} + self.metadata = {"login": "tenantuser", "email": "user@example.test", + "env": [{"key": "AUTH", "value": "uuid:key"}]} + self.couch = Mock() + self.couch.find_doc.return_value = {"docs": [self.metadata]} + self.kube = Mock() + self.kube.get_whisk_user.return_value = self.resource + self.service = AuthService(environ={}, couch_db=self.couch, kube_client=self.kube) + + def login(self, claims=None, **kwargs): + with patch("openserverless.impl.auth.auth_service.OidcTokenValidator") as validator: + validator.return_value.validate.return_value = claims if claims is not None else self.claims + with app.app_context(): + return self.service.login_oidc("signed-token", **kwargs) + + def denied(self, response, status=403): + self.assertEqual(status, response.status_code) + self.assertNotIn("AUTH", response.json) + self.kube.update_whisk_user.assert_not_called() + + def test_matching_immutable_identity_receives_own_credentials(self): + response = self.login() + self.assertEqual(200, response.status_code) + self.assertEqual("uuid:key", response.json["AUTH"]) + + def test_legacy_namespace_without_binding_is_denied(self): + self.resource["metadata"]["annotations"] = {} + self.denied(self.login()) + self.couch.find_doc.assert_not_called() + self.kube.create_whisk_user.assert_not_called() + + def test_different_identity_same_username_is_denied(self): + self.denied(self.login({**self.claims, "sub": "different-owner"})) + + def test_different_bound_issuer_is_denied(self): + self.resource["metadata"]["annotations"]["openserverless.apache.org/sso-issuer"] = "https://other.test" + self.denied(self.login()) + + def test_missing_and_non_string_identity_claims_are_denied(self): + for key in ("iss", "sub"): + for value in (None, "", " ", 123, ["owner-id"]): + with self.subTest(key=key, value=value): + self.denied(self.login({**self.claims, key: value}), 401) + + def test_partial_binding_is_denied(self): + for key in ("openserverless.apache.org/sso-issuer", "openserverless.apache.org/sso-sub"): + resource = copy.deepcopy(self.resource) + del resource["metadata"]["annotations"][key] + self.kube.get_whisk_user.return_value = resource + with self.subTest(key=key): + self.denied(self.login()) + + def test_disabled_matching_binding_is_denied(self): + self.resource["metadata"]["annotations"]["openserverless.apache.org/sso-disabled"] = "true" + self.denied(self.login()) + + def test_explicit_namespace_claim_does_not_bypass_binding(self): + self.service._environ = {"OIDC_NAMESPACE_CLAIM": "workspace"} + self.denied(self.login({**self.claims, "preferred_username": "attackeruser", + "workspace": "tenantuser", "sub": "attacker-id"})) + + def test_requested_namespace_does_not_bypass_binding(self): + self.denied(self.login({**self.claims, "sub": "attacker-id"}, expected_namespace="tenantuser")) + + def test_missing_whisk_user_does_not_expose_orphan_metadata(self): + self.kube.get_whisk_user.return_value = None + self.denied(self.login()) + + def test_revoked_binding_after_metadata_lookup_is_denied(self): + revoked = copy.deepcopy(self.resource) + revoked["metadata"]["annotations"]["openserverless.apache.org/sso-disabled"] = "true" + self.kube.get_whisk_user.side_effect = [self.resource, revoked] + self.denied(self.login()) + + def test_stale_credentials_after_recreation_are_not_returned(self): + recreated = copy.deepcopy(self.resource) + recreated["metadata"]["uid"] = "recreated-tenant-uid" + self.kube.get_whisk_user.side_effect = [self.resource, recreated] + self.denied(self.login()) + + def test_existing_metadata_credentials_remain_authoritative(self): + self.resource["spec"]["auth"] = "old-spec:old-key" + response = self.login() + self.assertEqual(200, response.status_code) + self.assertEqual("uuid:key", response.json["AUTH"]) + + def test_existing_resource_during_provisioning_must_match_identity(self): + self.resource["metadata"]["annotations"]["openserverless.apache.org/sso-sub"] = "other-id" + self.assertFalse(self.service.ensure_whisk_user("tenantuser", "tenantuser", "u@example.test", self.claims)) + self.kube.create_whisk_user.assert_not_called() + + def test_create_conflict_rejects_other_owner_and_legacy_and_disabled(self): + variants = [copy.deepcopy(self.resource) for _ in range(3)] + variants[0]["metadata"]["annotations"]["openserverless.apache.org/sso-sub"] = "other-id" + variants[1]["metadata"]["annotations"] = {} + variants[2]["metadata"]["annotations"]["openserverless.apache.org/sso-disabled"] = "true" + for resource in variants: + with self.subTest(resource=resource): + self.kube.get_whisk_user.side_effect = [None, resource] + self.kube.create_whisk_user.return_value = False + self.assertFalse(self.service.ensure_whisk_user("tenantuser", "tenantuser", "u@example.test", self.claims)) + + def test_create_conflict_accepts_same_bound_owner(self): + self.kube.get_whisk_user.side_effect = [None, self.resource] + self.kube.create_whisk_user.return_value = False + self.assertTrue(self.service.ensure_whisk_user("tenantuser", "tenantuser", "u@example.test", self.claims)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_user_provisioning.py b/tests/test_user_provisioning.py new file mode 100644 index 0000000..d3c81bd --- /dev/null +++ b/tests/test_user_provisioning.py @@ -0,0 +1,543 @@ +# 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 copy +from concurrent.futures import ThreadPoolExecutor +import io +import logging +import threading +import time +import unittest +from unittest.mock import Mock, patch + +import requests +from openserverless import app +from openserverless.impl.provisioning.kube_client import ProvisioningError, ProvisioningKubeClient +from openserverless.impl.provisioning.admission import ProvisioningAuthorizer +from openserverless.impl.provisioning.namespace_guard import NamespaceGuard +from openserverless.impl.provisioning.routes import route_context, tenant_routes +from openserverless.impl.provisioning.user_service import UserProvisioningService, SERVICE_NAMES + +TOKEN = "synthetic-provisioning-token-only" +BODY = {"user": "labtenant", "password": "Synthetic-pass!42", "email": "lab@example.test"} + + +class FakeKube: + def __init__(self, components=None): + self.secret = {"type": "Opaque", "data": {"token": base64.b64encode(TOKEN.encode()).decode()}} + self.configuration = {"spec": {"components": components if components is not None else + {name: True for name in SERVICE_NAMES if name != "minio"}}} + self.ingress = {"spec": {"ingressClassName": "traefik", "rules": [{"host": "example.test"}]}} + self.users = {} + self.routes = [] + self.lock = threading.Lock() + self.secret_reads = 0 + + def provisioning_secret(self): + self.secret_reads += 1 + return self.secret + + def existing_user(self, name): + return copy.deepcopy(self.users.get(name)) + + def cluster(self): return self.configuration + def apihost_ingress(self): return self.ingress + def streamer_service(self): return {"metadata": {"name": "openserverless-streamer-api"}} + + def create_user(self, body): + with self.lock: + name = body["metadata"]["name"] + if name in self.users: + raise ProvisioningError("Resource already exists", 409) + result = copy.deepcopy(body) + result["metadata"]["uid"] = f"synthetic-{name}-uid" + self.users[name] = result + return copy.deepcopy(result) + + def create_route(self, body): self.routes.append(copy.deepcopy(body)) + + +class ProvisioningApiTest(unittest.TestCase): + def setUp(self): + self.kube = FakeKube() + self.factory = patch("openserverless.rest.users.ProvisioningKubeClient", return_value=self.kube) + self.factory.start() + self.addCleanup(self.factory.stop) + # Service-contract tests refresh synchronously. Actual asynchronous + # refresh, expiry and invalid/valid contention have focused tests below. + self.authorizer = ProvisioningAuthorizer(loader=self.kube.provisioning_secret) + self.authorizer_patch = patch("openserverless.rest.users.provisioning_authorizer", self.authorizer) + self.authorizer_patch.start() + self.addCleanup(self.authorizer_patch.stop) + self.refresh_patch = patch.object(self.authorizer, "start", self.authorizer._refresh_once) + self.refresh_patch.start() + self.addCleanup(self.refresh_patch.stop) + self.guard_patch = patch("openserverless.impl.provisioning.user_service.NamespaceGuard", autospec=True) + self.guard = self.guard_patch.start().return_value + self.addCleanup(self.guard_patch.stop) + + def post(self, body=BODY, token=TOKEN, **kwargs): + headers = {} if token is None else {"Authorization": "Bearer " + token} + with app.test_client() as client: + return client.post("/system/api/v1/users", headers=headers, json=body, **kwargs) + + def test_defaults_create_full_tenant_and_owned_routes_without_credentials_in_response(self): + response = self.post() + self.assertEqual(202, response.status_code) + self.assertEqual("pending", response.json["provisioning_status"]) + self.assertEqual([s for s in SERVICE_NAMES if s != "minio"], response.json["services"]) + resource = self.kube.users[BODY["user"]] + spec = resource["spec"] + self.assertEqual(BODY["password"], spec["password"]) + self.assertNotIn("annotations", resource["metadata"]) + self.assertEqual("labtenant", spec["mongodb"]["database"]) + self.assertEqual("labtenant", spec["redis"]["prefix"]) + self.assertEqual("auto", spec["object-storage"]["quota"]) + self.assertTrue(spec["object-storage"]["data"]["enabled"]) + self.assertTrue(spec["object-storage"]["route"]["enabled"]) + self.assertEqual(6, len(self.kube.routes)) + for route in self.kube.routes: + self.assertEqual(resource["metadata"]["uid"], route["metadata"]["ownerReferences"][0]["uid"]) + for secret in (TOKEN, spec["auth"], spec["password"], spec["redis"]["password"]): + self.assertNotIn(secret, response.get_data(as_text=True)) + + def test_slim_defaults_enable_only_installed_services(self): + self.kube.configuration["spec"]["components"] = {"redis": True, "seaweedfs": True} + response = self.post() + self.assertEqual(202, response.status_code) + self.assertEqual(["redis", "seaweedfs"], response.json["services"]) + self.assertNotIn("mongodb", self.kube.users["labtenant"]["spec"]) + + def test_explicit_subset_and_per_request_credentials_are_isolated(self): + self.assertEqual(202, self.post({**BODY, "services": "postgres, redis"}).status_code) + second = self.post({**BODY, "user": "seconduser", "services": "mongodb"}) + self.assertEqual(202, second.status_code) + first = self.kube.users["labtenant"]["spec"] + other = self.kube.users["seconduser"]["spec"] + self.assertNotIn("mongodb", first) + self.assertNotIn("redis", other) + self.assertNotEqual(first["auth"], other["auth"]) + self.assertNotIn("object-storage", first) + + def test_unavailable_unknown_or_ambiguous_services_never_create(self): + self.kube.configuration["spec"]["components"] = {"redis": True} + for services in ("mongodb", "unrecognized", "", "all,redis", "redis,redis", "redis,", None, [], "minio"): + with self.subTest(services=services): + self.assertEqual(400, self.post({**BODY, "services": services}).status_code) + self.assertEqual({}, self.kube.users) + self.assertEqual([], self.kube.routes) + + def test_backend_name_limits_fail_before_creation(self): + for name, services in (("a" * 55, "mongodb"), ("a" * 59, "seaweedfs")): + self.assertEqual(400, self.post({**BODY, "user": name, "services": services}).status_code) + self.assertEqual({}, self.kube.users) + self.assertEqual(202, self.post({**BODY, "user": "a" * 61, "services": "redis"}).status_code) + + def test_no_auth_or_invalid_auth_never_checks_or_creates_users(self): + with patch.object(self.kube, "existing_user") as lookup: + for token in (None, "wrong-token", "namespace-uuid:tenant-key"): + self.assertEqual(401, self.post(token=token).status_code) + for header in ("Basic abc", "Bearer", "Bearer a b", "Bearer \u00e9", "Bearer " + "a" * 8193): + with app.test_client() as client: + self.assertEqual(401, client.post("/system/api/v1/users", headers={"Authorization": header}, json=BODY).status_code) + lookup.assert_not_called() + self.assertEqual({}, self.kube.users) + + def test_missing_malformed_and_unreadable_secret_fail_closed(self): + for secret in (None, [], {}, {"type": "Opaque", "data": {}}, {"type": "Opaque", "data": {"token": "%%%"}}, + {"type": "Opaque", "data": {"token": ""}}, {"type": "Opaque", "data": {"token": "IA=="}}, + {"type": "kubernetes.io/tls", "data": {"token": "dG9rZW4="}}): + self.kube.secret = secret + self.assertEqual(503, self.post().status_code) + with patch.object(self.kube, "provisioning_secret", side_effect=ProvisioningError("dependency unavailable")): + self.assertEqual(503, self.post().status_code) + self.assertEqual({}, self.kube.users) + + def test_rotation_revocation_without_restart(self): + self.assertEqual(202, self.post().status_code) + self.kube.secret["data"]["token"] = base64.b64encode(b"rotated-synthetic").decode() + self.assertEqual(401, self.post({**BODY, "user": "seconduser"}).status_code) + self.assertEqual(202, self.post({**BODY, "user": "seconduser"}, token="rotated-synthetic").status_code) + self.kube.secret = None + self.assertEqual(503, self.post({**BODY, "user": "thirduser"}, token="rotated-synthetic").status_code) + self.assertEqual(4, self.kube.secret_reads) + + def test_existing_blocked_deleting_users_are_unchanged(self): + self.post() + self.kube.users["labtenant"]["spec"]["blocked"] = True + self.kube.users["labtenant"]["metadata"]["deletionTimestamp"] = "2026-10-21T00:00:00Z" + before = copy.deepcopy(self.kube.users) + routes = copy.deepcopy(self.kube.routes) + self.assertEqual(409, self.post({**BODY, "password": "different", "services": "redis"}).status_code) + self.assertEqual(before, self.kube.users) + self.assertEqual(routes, self.kube.routes) + + def test_retained_namespace_conflict_and_unverifiable_state_do_not_create(self): + for code in (409, 503): + self.guard.assert_unused.side_effect = ProvisioningError("Namespace unavailable", code) + self.assertEqual(code, self.post().status_code) + self.assertEqual({}, self.kube.users) + self.assertEqual([], self.kube.routes) + + def test_invalid_bearers_cannot_lock_out_valid_requests_or_trigger_secret_reads(self): + now = [0.0] + auth = ProvisioningAuthorizer(loader=self.kube.provisioning_secret, clock=lambda: now[0]) + with patch("openserverless.rest.users.provisioning_authorizer", auth), patch.object(auth, "start"): + for cycle in range(6): + now[0] = float(cycle) + auth._refresh_once() + for index in range(20): + self.assertEqual(401, self.post(token="invalid-" + str(index)).status_code) + self.assertEqual(202, self.post({**BODY, "user": "labtenant" + str(cycle)}).status_code) + self.assertEqual(6, self.kube.secret_reads) + + def test_slow_refresh_keeps_valid_and_invalid_requests_independent(self): + entered, release = threading.Event(), threading.Event() + now = [0.0] + auth = ProvisioningAuthorizer(loader=self.kube.provisioning_secret, clock=lambda: now[0]) + auth._refresh_once() + def slow_secret(): + entered.set() + if not release.wait(5): + raise RuntimeError("test synchronization timed out") + return self.kube.secret + with patch("openserverless.rest.users.provisioning_authorizer", auth), patch.object(auth, "start"), \ + patch.object(self.kube, "provisioning_secret", side_effect=slow_secret) as lookup: + auth._loader = self.kube.provisioning_secret + with ThreadPoolExecutor(max_workers=4) as pool: + first = pool.submit(auth._refresh_once) + try: + self.assertTrue(entered.wait(2)) + other = list(pool.map(lambda _: self.post(token="invalid").status_code, range(12))) + self.assertEqual([401] * 12, other) + self.assertEqual(202, self.post().status_code) + lookup.assert_called_once() + with app.test_client() as client: + self.assertEqual(200, client.get("/system/apispec_1.json").status_code) + now[0] = 2.0 + self.assertEqual(503, self.post().status_code) + finally: + release.set() + first.result(timeout=2) + + def test_dependency_failure_revokes_snapshot_and_refresh_recovers(self): + auth = ProvisioningAuthorizer(loader=self.kube.provisioning_secret) + auth._refresh_once() + with patch("openserverless.rest.users.provisioning_authorizer", auth), patch.object(auth, "start"): + with patch.object(auth, "_loader", side_effect=RuntimeError("sensitive diagnostics")) as lookup: + auth._refresh_once() + response = self.post() + self.assertEqual(503, response.status_code) + self.assertEqual("1", response.headers["Retry-After"]) + self.assertNotIn("sensitive", response.get_data(as_text=True)) + self.assertEqual(503, self.post().status_code) + lookup.assert_called_once() + auth._refresh_once() + self.assertEqual(202, self.post().status_code) + + def test_concurrent_creation_returns_one_acceptance_one_conflict(self): + barrier = threading.Barrier(2) + def lookup(name): + barrier.wait(timeout=5) + return None + with patch.object(self.kube, "existing_user", side_effect=lookup): + with ThreadPoolExecutor(max_workers=2) as executor: + responses = list(executor.map(lambda _: self.post().status_code, range(2))) + self.assertEqual([202, 409], sorted(responses)) + self.assertEqual(1, len(self.kube.users)) + self.assertEqual(6, len(self.kube.routes)) + + def test_validation_payload_types_unknown_fields_and_no_normalization(self): + invalid = [None, [], "text", {}, {**BODY, "role": "admin"}, {**BODY, "token": TOKEN}] + for field in ("user", "password", "email"): + invalid += [{k: v for k, v in BODY.items() if k != field}] + invalid += [{**BODY, field: value} for value in (None, "", " ", 123, True, [], "a\nb")] + invalid += [{**BODY, "user": value} for value in ("ABcde", "abc", "a-bcd", "a/bcd", "openserverless", "a" * 62)] + invalid += [{**BODY, "email": value} for value in ("a@b", "a b@example.test", "a@@b.test", 'a"@example.test', 'a\\b@example.test', "a..b@example.test", "a@-example.test", "a@exa_mple.test")] + invalid += [{**BODY, "password": "\ud800"}, {**BODY, "email": "\ud800@example.test"}, {**BODY, "password": "x" * 73}, {**BODY, "password": "\u00e9" * 37}] + with app.test_client() as client: + for body in invalid: + with self.subTest(body_type=type(body).__name__): + # data=json is explicit here because Flask json=None omits the body. + import json + response = client.post("/system/api/v1/users", data=json.dumps(body), content_type="application/json", + headers={"Authorization": "Bearer " + TOKEN}) + self.assertEqual(400, response.status_code) + self.assertEqual({}, self.kube.users) + + def test_content_type_json_errors_body_limit_and_auth_first(self): + with app.test_client() as client: + for data, content_type, code in (("{", "application/json", 400), ("{}", "text/plain", 415), + (" " * 16385, "application/json", 413)): + response = client.post("/system/api/v1/users", data=data, content_type=content_type, + headers={"Authorization": "Bearer " + TOKEN}) + self.assertEqual(code, response.status_code) + self.assertEqual(401, client.post("/system/api/v1/users", data="{", content_type="text/plain").status_code) + + def test_dependency_lookup_failure_is_not_treated_as_absent(self): + with patch.object(self.kube, "existing_user", side_effect=ProvisioningError("dependency unavailable")): + self.assertEqual(503, self.post().status_code) + self.assertEqual({}, self.kube.users) + + def test_partial_route_failure_is_explicit_and_does_not_delete_user(self): + with patch.object(self.kube, "create_route", side_effect=ProvisioningError("dependency unavailable")): + response = self.post() + self.assertEqual(503, response.status_code) + self.assertIn("User created but route setup failed", response.json["message"]) + self.assertIn("labtenant", self.kube.users) + self.assertEqual(409, self.post().status_code) + + def test_config_and_ingress_failure_no_user_created(self): + for config in ({}, {"spec": {"components": {"redis": "false"}}}): + self.kube.configuration = config + self.assertEqual(503, self.post().status_code) + self.kube.configuration = {"spec": {"components": {}}} + self.kube.ingress["spec"]["ingressClassName"] = "unsupported" + self.assertEqual(503, self.post().status_code) + self.assertEqual({}, self.kube.users) + + def test_no_sensitive_error_or_request_logs(self): + log = io.StringIO() + handler = logging.StreamHandler(log) + logging.getLogger().addHandler(handler) + self.addCleanup(logging.getLogger().removeHandler, handler) + with patch.object(self.kube, "create_user", side_effect=RuntimeError(BODY["password"] + TOKEN)): + response = self.post() + self.assertEqual(500, response.status_code) + for value in (TOKEN, BODY["password"]): + self.assertNotIn(value, log.getvalue() + response.get_data(as_text=True)) + + def test_spec_exposes_contract_without_real_credentials(self): + with app.test_client() as client: + response = client.get("/system/apispec_1.json") + self.assertEqual(200, response.status_code) + spec = response.json + self.assertIn("/system/api/v1/users", spec["paths"]) + self.assertIn("provisioningBearer", spec["securityDefinitions"]) + + +class ProvisioningKubeClientTest(unittest.TestCase): + def setUp(self): + self.client = ProvisioningKubeClient.__new__(ProvisioningKubeClient) + self.client.host = "https://kubernetes.test" + self.client.token = "Bearer synthetic-service-account" + self.client.ssl_ca_cert = "/synthetic/ca.crt" + + @patch("openserverless.impl.provisioning.kube_client.requests.request") + def test_missing_vs_forbidden_error_conflict_json_and_transport(self, request): + response = request.return_value + response.status_code = 404 + self.assertIsNone(self.client.existing_user("validuser")) + for code in (401, 403, 500, 503): + response.status_code = code + with self.assertRaises(ProvisioningError) as error: self.client.existing_user("validuser") + self.assertEqual(503, error.exception.status) + response.status_code = 409 + with self.assertRaises(ProvisioningError) as error: self.client.create_user({}) + self.assertEqual(409, error.exception.status) + response.status_code = 201 + response.json.return_value = {"metadata": {"uid": "uid"}} + self.assertEqual("uid", self.client.create_user({})["metadata"]["uid"]) + self.assertEqual((5, 20), request.call_args.kwargs["timeout"]) + self.assertEqual("/synthetic/ca.crt", request.call_args.kwargs["verify"]) + response.json.side_effect = ValueError("sensitive response") + with self.assertRaises(ProvisioningError): self.client.create_user({}) + request.side_effect = requests.Timeout("sensitive request") + with self.assertRaises(ProvisioningError) as error: self.client.create_user({}) + self.assertNotIn("sensitive", str(error.exception)) + + +class ProvisioningAuthorizerTest(unittest.TestCase): + def test_cold_start_and_expired_snapshot_fail_closed(self): + now = [0.0] + kube = FakeKube() + auth = ProvisioningAuthorizer(loader=kube.provisioning_secret, clock=lambda: now[0]) + with patch.object(auth, "start"): + with self.assertRaises(ProvisioningError) as error: + auth.authorize(TOKEN.encode()) + self.assertEqual(503, error.exception.status) + auth._refresh_once() + auth.authorize(TOKEN.encode()) + now[0] = 2.0 + with self.assertRaises(ProvisioningError) as error: + auth.authorize(TOKEN.encode()) + self.assertEqual(503, error.exception.status) + self.assertEqual(1, kube.secret_reads) + + def test_slow_response_cannot_extend_revoked_token_lifetime(self): + now = [0.0] + def slow_loader(): + secret = FakeKube().secret + now[0] += 2.0 + return secret + auth = ProvisioningAuthorizer(loader=slow_loader, clock=lambda: now[0]) + with patch.object(auth, "start"): + auth._refresh_once() + with self.assertRaises(ProvisioningError) as error: + auth.authorize(TOKEN.encode()) + self.assertEqual(503, error.exception.status) + + def test_rotation_revocation_and_recovery_do_not_depend_on_client_traffic(self): + kube = FakeKube() + auth = ProvisioningAuthorizer(loader=kube.provisioning_secret) + with patch.object(auth, "start"): + auth._refresh_once() + auth.authorize(TOKEN.encode()) + kube.secret["data"]["token"] = base64.b64encode(b"rotated-synthetic").decode() + auth._refresh_once() + with self.assertRaises(ProvisioningError) as error: + auth.authorize(TOKEN.encode()) + self.assertEqual(401, error.exception.status) + auth.authorize(b"rotated-synthetic") + kube.secret = None + auth._refresh_once() + with self.assertRaises(ProvisioningError) as error: + auth.authorize(b"rotated-synthetic") + self.assertEqual(503, error.exception.status) + kube.secret = FakeKube().secret + auth._refresh_once() + auth.authorize(TOKEN.encode()) + self.assertEqual(4, kube.secret_reads) + + def test_only_one_background_worker_under_concurrent_invalid_and_valid_requests(self): + kube = FakeKube() + auth = ProvisioningAuthorizer(loader=kube.provisioning_secret) + self.addCleanup(auth.close) + entered, release = threading.Event(), threading.Event() + def blocked_loader(): + entered.set() + if not release.wait(3): + raise RuntimeError("test synchronization timed out") + return kube.provisioning_secret() + auth._loader = blocked_loader + try: + with ThreadPoolExecutor(max_workers=16) as pool: + list(pool.map(lambda _: auth.start(), range(32))) + self.assertTrue(entered.wait(1)) + worker = auth._thread + with self.assertRaises(ProvisioningError) as error: + auth.authorize(TOKEN.encode()) + self.assertEqual(503, error.exception.status) + self.assertIs(worker, auth._thread) + release.set() + deadline = time.monotonic() + 1 + while True: + try: + auth.authorize(TOKEN.encode()) + break + except ProvisioningError: + if time.monotonic() > deadline: + self.fail("background refresh never became ready") + time.sleep(0.01) + def request(index): + try: + auth.authorize(TOKEN.encode() if index % 2 else b"invalid-" + str(index).encode()) + return 200 + except ProvisioningError as error: + return error.status + with ThreadPoolExecutor(max_workers=16) as pool: + results = list(pool.map(request, range(200))) + self.assertEqual([401, 200] * 100, results) + self.assertEqual(1, kube.secret_reads) + finally: + release.set() + auth.close() + if auth._thread is not None: + auth._thread.join(timeout=1) + + +class NamespaceGuardTest(unittest.TestCase): + def setUp(self): + self.patch = patch("openserverless.impl.provisioning.namespace_guard.CouchDB") + self.db = self.patch.start().return_value + self.addCleanup(self.patch.stop) + self.db.db_base = "http://couchdb.test/openserverless_" + self.get = self.db.db_session.get + self.response = self.get.return_value + self.response.status_code = 200 + + def rows(self, *bodies): + self.response.json.side_effect = [{"rows": rows} for rows in bodies] + + def test_fresh_name_checks_identity_entities_and_activation_history(self): + self.rows([{"key": "labtenant", "error": "not_found"}], [], []) + NamespaceGuard().assert_unused("labtenant") + self.assertEqual(3, self.get.call_count) + calls = self.get.call_args_list + self.assertEqual('["labtenant"]', calls[0].kwargs["params"]["keys"]) + for call in calls[1:]: + self.assertEqual('"labtenant/"', call.kwargs["params"]["startkey"]) + self.assertEqual('"labtenant0"', call.kwargs["params"]["endkey"]) + self.assertEqual("false", call.kwargs["params"]["inclusive_end"]) + self.assertEqual(1, call.kwargs["params"]["limit"]) + self.assertEqual((2, 3), call.kwargs["timeout"]) + self.db.db_session.close.assert_called_once() + + def test_live_or_deleted_subject_and_orphaned_package_or_activation_conflict(self): + cases = [ + ([{"key": "labtenant", "id": "labtenant", "value": {"rev": "1-live"}}],), + ([{"key": "labtenant", "id": "labtenant", "value": {"rev": "2-gone", "deleted": True}}],), + ([{"key": "labtenant", "error": "not_found"}], [{"id": "labtenant/pkg/action"}]), + ([{"key": "labtenant", "error": "not_found"}], [], [{"id": "labtenant/activation"}]), + ] + for bodies in cases: + with self.subTest(bodies=bodies): + self.rows(*bodies) + with self.assertRaises(ProvisioningError) as error: + NamespaceGuard().assert_unused("labtenant") + self.assertEqual(409, error.exception.status) + + def test_malformed_forbidden_missing_database_and_transport_fail_closed(self): + for data in ({}, [], {"rows": None}, {"rows": []}, {"rows": [{"key": "other", "error": "not_found"}]}): + self.response.json.side_effect = None + self.response.json.return_value = data + with self.assertRaises(ProvisioningError) as error: + NamespaceGuard().assert_unused("labtenant") + self.assertEqual(503, error.exception.status) + for code in (401, 403, 404, 500): + self.response.status_code = code + with self.assertRaises(ProvisioningError) as error: + NamespaceGuard().assert_unused("labtenant") + self.assertEqual(503, error.exception.status) + self.get.side_effect = requests.Timeout("sensitive diagnostics") + with self.assertRaises(ProvisioningError) as error: + NamespaceGuard().assert_unused("labtenant") + self.assertNotIn("sensitive", str(error.exception)) + + +class TenantRouteTest(unittest.TestCase): + def test_nginx_traefik_tls_ownership_and_sigv4_path_contracts(self): + for ingress_class in ("nginx", "traefik"): + for tls in (True, False): + routes = tenant_routes("labtenant", (ingress_class, "example.test", tls), ["seaweedfs"], True, "owner-uid") + ingresses = [r for r in routes if r["kind"] == "Ingress"] + self.assertEqual(4, len(ingresses)) + for ingress in ingresses: + self.assertEqual(tls, bool(ingress["spec"].get("tls"))) + self.assertEqual("owner-uid", ingress["metadata"]["ownerReferences"][0]["uid"]) + s3 = ingresses[-1] + self.assertEqual(["/labtenant-web", "/labtenant-data"], [p["path"] for p in s3["spec"]["rules"][0]["http"]["paths"]]) + self.assertFalse(any("rewrite" in key or "middlewares" in key for key in s3["metadata"]["annotations"])) + serialized = str(routes) + self.assertIn("/web/labtenant/$2", serialized) + self.assertIn("/action/labtenant/$2", serialized) + self.assertEqual([], tenant_routes("labtenant", ("nginx", "example.test", False), ["redis"], False, "uid")) + + +if __name__ == "__main__": unittest.main()