From 15daa4f47eaa5a9b591b6bc3b8aff3f80a3763a8 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 19 Oct 2026 14:31:39 +0200 Subject: [PATCH 1/3] Isolate etcd and Milvus credentials and management endpoints on 0.9.0 --- .github/workflows/check.yml | 2 + deploy/etcd/etcd-jwt-secret.yaml | 27 --- deploy/etcd/etcd-sts.yaml | 2 +- deploy/milvus-slim/milvus-cfg-slim-base.yaml | 7 +- deploy/milvus-slim/milvus-svc.yaml | 4 +- deploy/milvus-slim/milvus.yaml | 5 +- deploy/milvus/milvus-cfg-base.yaml | 7 +- deploy/milvus/milvus-svc.yaml | 4 +- deploy/milvus/milvus.yaml | 5 +- .../operator-roles.yaml | 5 + openserverless/config.py | 2 +- openserverless/datastore_security.py | 108 +++++++++++ openserverless/etcd.py | 83 +------- openserverless/etcd_admin.py | 113 +++++++++++ openserverless/milvus_standalone.py | 33 +++- openserverless/templates/milvus-cfg-base.yaml | 15 +- .../templates/milvus-cfg-slim-base.yaml | 15 +- tests/test_datastore_security.py | 178 ++++++++++++++++++ 18 files changed, 477 insertions(+), 138 deletions(-) delete mode 100644 deploy/etcd/etcd-jwt-secret.yaml create mode 100644 openserverless/datastore_security.py create mode 100644 openserverless/etcd_admin.py create mode 100644 tests/test_datastore_security.py diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 1c54bf2..205f79e 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -64,6 +64,8 @@ jobs: run: bash .github/cisetup.sh - name: Install run: task setup + - name: Datastore security regression tests + run: poetry run python3 -m unittest discover -s tests -p test_datastore_security.py -v - name: BuildAndLoad run: task build-and-load - name: AddIngress diff --git a/deploy/etcd/etcd-jwt-secret.yaml b/deploy/etcd/etcd-jwt-secret.yaml deleted file mode 100644 index 0841f86..0000000 --- a/deploy/etcd/etcd-jwt-secret.yaml +++ /dev/null @@ -1,27 +0,0 @@ -# -# 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. -# ---- -apiVersion: v1 -kind: Secret -metadata: - name: openserverless-etcd-jwt-token - namespace: openserverless - labels: - app.kubernetes.io/name: openserverless-etcd -type: Opaque -data: - jwt-token.pem: "LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlKS1FJQkFBS0NBZ0VBdXlNcjdmOUJWajJ2RjhNUC8zNStrcnA0MlF0Nk5YMmEvS3IyY24xNjBSajBuZ0lvCk1qL1RmdXVnMmJnaHlEdnhDdytKZU1QUUNYMXRGZENzM2RHZ0xmSFppM0MyR21VV250VG1WQTRCSENZdmc4RisKRnBYQURjdE5EKzRWc1F4OUVsN1pKR0lxWFkySThFM3JPWkd4SnhxQmR1bDIwVy9xczNSNFFRMlFsbzU0dzNhagpmVWZCQ3FoeTdveXJWclN5NFgwN0JYL09CYkRCU2NBNUJlTG9ZOFdTNWwvYWFvZnRiVlpqYmxHYkE5S0tvclFNCmN6QjNxenBzQXZyUjZkWVBES1dFMVpGQ1huWjBBdjBsQlZMckk4S1ZkREFaWHBRelFEUDl5dytCMWpaelhiaG0KRGg5UmN6YUt3dXJ5UTA4d0JVanpUU09ydkdaY254aEJQUjZ5dTUwUzYranJEZGU0b2J0anpYTHZGQkFmdXNLSQpLczZtRG1sYUpKVGVybUlQNnFZOHZKRUx6eTNYYk40R2tyNk5NMjZFdkxnTnRXYmhtSVBzRysyRVBTelErV3JaCmFsTFNVcFRSU0kzRjl5THFrN21ybE9ZZUxxTjRoYU1rem4ySFc2RXhZL25kSjJJMXBVUWZIQ3d5Y0MxMDNhSDUKN29HZ01DTGpWdnBENjFMZ3hXa3U3dnRGcjI0L1lpS3B1SGc5V21UQlNKY05CcVJ4b3l5UFlRRWJsNjBFLy80UgpLc01XenJLdm9qeFlXQ1ZsLzV1ZElnUGoraDM1VkZ6L3VxOExvLy83QkZ6WWRqRWhLZXc0cHBPc2xIMXlkY3FsCmNPRHNQQUg4L1UxbFZpUzFaY0FRV3V6QWo2b3ppRWlrd3F1RnFYMHdOQnNURkt2YWxiNmYvUFBHc2FVQ0F3RUEKQVFLQ0FnQktVWmFCcHl5Z2ZsREtCQXFYY2ZuOGRaZkd3cnNmQVNvTERDNWhtNW1wN3RvKzRpWnFobFNOQkREVwpKajBzVXpuUUpiUjN4SFFiUUxlUXdWWFZwZ21TRllsS3lsRlh2Q3UybW1CNm96RW5wN2FaQWIvOFlpNXFqQUduCndPcldiRnBEY0cyVitISlF3d1BPUTNPV1lsalRhY0JlcENzdTA5Nkd0RXVKc2tSb3RsNFpha1ZWWEtpNmZkbGYKQ1hpR0w2OFd3L0FycCt6Kzc0cGpxclk2c3ptTlhncVArU001eVkrdUsyL0RnUjhwN25SbVYxRFdjam9rUTBoZwpDcEpRNWxVRnRicENRclR6NU9EL3l0N245VjB5ejNVaE5uVHN6YmhvMS8wRHByM3VmNk5qWmZjSnhzQkhIRE1SClk5TWdqaE5KRmdhaDdQZW9FNUxYRC8vZ3NROVRxZHZmOGpsT2d1OWFzWGF6eGtiNGw4NTc2czVTNkJtbUN5QlIKTG1USGVTSGpvLzBVQ0psU0pWUTZsdDZNZ2FZRHVBMXNZZjhDdkdtT0doRjQzM1pyUWtqTkdSYnZPTVg2R3luegp1aDlvVWhndDFuaFVNSE5iaEZvNjZRbW1tNERmMVJudnpFbTVJVmk1MXY0dnE1eWI5OUFSd0dBQmg1S3V5d292Cm9QaU9OZndwRlF3WGRzWlo2cllDcFJjMUpVbHE0Q25CcnkxbTQ4eDRNNURmakxjaDVmelNLakg1UFd5ZnE2SkIKditPUzRSalZtMlljRThzQk5iVXcyYUF2TW5OQzNLWE5RbFhuV1IzVGg2T05veXhqc3lwN3ErTHNSdTd3RkZlWgoyUTFya21IQ3kyWlFsOFRSeUhObzZ1UXc1aTJ2SGZEWE5abm8vcmZwNTZuUEoxQk5vUUtDQVFFQTlzM0tMSVJlCjJpN0VoekZiRSsvRHNMRnJjeXhlWlJ5V2x4dXR1Q3BCZk1ET0tPTDhtYjhsT1MxcnRzTDZSSit0a0ZhYmFYeXkKTXJoM3dxaW5zazBBRVM5QlMyeU1YaU5PT3RSbjljZkRHMVpPcU1vOXYwVFJXRUZtUlFTQ0x1QThPeXlKZmZKcAorV2psTDNPZU5RTzB4Sk9pNVFxbzVEdDVwSnhvZEl0NXVlOHR1ZUVjaWt6Wm80aUF0Q3oweUovYWVpR3VyOWgwCkZyZlFHL1BGeGtWeVdia210U2lwM3NRbE03d29RZ3Y2ay9rV3NIQ1hZNjI0YzN4cytieThrczRLYzVlSGZxeXAKQUxNZ0NnNXFuVjBFbHdEWThCd1l1NlFETHZsdlF1cVNqNlB4NFRjY2I2NWFJUy9QUm5GOEtQbWZlV3kxaWs5Zgp2SmxSaWNQaWtWMS95UUtDQVFFQXdodzhZV3FLdG90R050UkdieVArVXRTT252OXhIR05rUjRzN0ZRYVltNGF0ClB4SjZlbkJGd3F5VkV6b1VuNkk1cGdhQnRHNzBDdHRQanR2R0lzVkZ2S3BGMGdRZSsrdkQwNHl6RW4rUHFBSWUKUkVCd0NmSWFKWTVWMFJwUHZRdnlrNDFDWG5rTlRsZXc0eEVtV1Zlc1l0U1Fwb3pwMllJeWlYcjhuWjY0a1hsKwp1cDBWdUc2VENPTllhbnhQUTFEbFViOCtzZlBtaCtqSC84U1o5d3ppK0pBQUlIT2VaVktWSFFGcEJPWDB1dU1BCngwR0VVbXN2QlJVeXRBM0RJenEraXRGMFdUeWRQNlR5UU1HaTJVamYrcHNObVhJODgxOVluSHlRaGVOeEZ4aVUKK2dpNDVXQ1JEbFEzWmIrdmZzRWo5S3o5QUl4N1lObmljNFI4Znlnby9RS0NBUUE1Q2JKUnUzM1lXcnFjcS9GcQptQnpYdHdrb0l2YmwwMGVTUUpsVFdLQ3QzamxIQmI1dklkZk9jUXJWd0JYcXlpUm9tVHR0azUzelR1S09vRCt4ClcwNUtxbXNpclVGN2VCbGFTMENEY3RsS2RoWFR1SG13dkl6TklzQXA3ekxPOW8rMDlVUUpVWDdnWHplb1VjWWMKWm5EZWhXZytHc3FSaEpWR1Q4ZHNOOWUzK3VqTWJwUVQrbExLRW4vOEkrK21kaHNkNHE3bWdOWERjbE9zUWhoUgpERGV6T0xFVGFRUnBNVmVQcjBwMmlTSXVGSUVENGVGNjdla1U5NHFQNU9pSDNRcFpVSW5GNHMxa0ZnR3ZrVWxlCmVPTFlxeDkzem1hQTV4UzlSSFJNT0w0S2FFODhSMnBRL1RCalB4TmVLbGdSZUM4ODlyeTE3RVlwNWgwTUE3TlcKK1E3UkFvSUJBUUNqa0pkd253NXo4N2dleTVEdVppNFFMZnhnRG0rSnRGYnowbWJldU1ndmhiVEd5Uk5ieU85LwpzMTJLN2c0aTFkZ3VCcDJWSlVtUFBHNDF1d3Vpa1FqS2lkQjd2KzA3NnRjQWxEYnhCM3hOM2pzUzJCbVF4K2t3CmJGZno4Mi90cFB1clNnVDdCcFlueGN6UUNjdCt6Vm45alpDMlIxZG1hVWVjZ1g0TWM0cWpTdlYyMnBkSzZpSU0KQTBhbHRFbzJOeHlGWGRtK3FmOC8xdTl3bm9IK28yVStmbGtjLzFBaG9RejFmQks2aGV0Szc2M2dPVnorOWtLUQp6ZVFEeFBVQjRxWmM0MUpmUDZ4aHpPUTFuUFZVa3d2cGFHNTFOc1lWYmVxRk1wRzF3SW16Skl1a2hXVUlJKzhYClcrK3lWV3ZVSzN2WTJVdjRBbWlVbkFVdHdoNEdLQmtoQW9JQkFRQ2gzeHk3Uk50QjUzREVUYndJbUtlUUxscGgKTXQzNWZLdWdwRTFDb1Y5WTNVZEpaczdlQ1YyN2pzcEFHZ3VxKytzODg1MEw4L3RuVGd5SHZWbVBlSzZGdW5GTgpEak9uNUNHOHByYVNmTVAvNTZjN282R29aRk1NUUtoZDBzOFczODN0Q2Jobk9YQng1S1FOa2hMNWQ2TlZrQzB2Cnd2WGpMUmdud2IxNlJnaGhESUQwVFhYNHc1UXlMVGVkTEYwZnc5Z0ZPeHY4VFhteWVzOGNVdzh5a3RnUEhMYTAKM1dFbzR1U1N4eE9zQTB4QzZ2dTZZM3VybWtVWTluYWpGUDcwaUtMT1dBWkozVkFQeTUxS2kzU2xNRHBRRmpCTgpmcXg2aXFvUGU2L0ZKc29xcHVoN29UNXEyRzhkMWdWa3h1ak1MQXZ5Z2pHQ2JtczVVV3ZFNS9TQ2hvMU0KLS0tLS1FTkQgUlNBIFBSSVZBVEUgS0VZLS0tLS0K" \ No newline at end of file diff --git a/deploy/etcd/etcd-sts.yaml b/deploy/etcd/etcd-sts.yaml index f3d915f..1e828b6 100644 --- a/deploy/etcd/etcd-sts.yaml +++ b/deploy/etcd/etcd-sts.yaml @@ -152,7 +152,7 @@ spec: - name: etcd-jwt-token secret: defaultMode: 256 - secretName: openserverless-etcd-jwt-token + secretName: openserverless-etcd-jwt-token-v2 volumeClaimTemplates: - metadata: name: openserverless-etc-data diff --git a/deploy/milvus-slim/milvus-cfg-slim-base.yaml b/deploy/milvus-slim/milvus-cfg-slim-base.yaml index fd83d22..03e2b90 100644 --- a/deploy/milvus-slim/milvus-cfg-slim-base.yaml +++ b/deploy/milvus-slim/milvus-cfg-slim-base.yaml @@ -17,11 +17,12 @@ # --- apiVersion: v1 -kind: ConfigMap +kind: Secret metadata: name: openserverless-milvus namespace: openserverless -data: +type: Opaque +stringData: default.yaml: |+ # Copyright (C) 2019-2021 Zilliz. All rights reserved. # @@ -38,7 +39,7 @@ data: rootPath: milvus auth: enabled: true - userName: root + userName: etcdmilvus password: 0therPa55 endpoints: - openserverless-etcd-0.openserverless-etcd-headless.openserverless.svc.cluster.local:2379 diff --git a/deploy/milvus-slim/milvus-svc.yaml b/deploy/milvus-slim/milvus-svc.yaml index 30e687a..e3f35a1 100644 --- a/deploy/milvus-slim/milvus-svc.yaml +++ b/deploy/milvus-slim/milvus-svc.yaml @@ -26,17 +26,15 @@ metadata: app.kubernetes.io/instance: openserverless-milvus component: "standalone" spec: - type: NodePort + type: ClusterIP ports: - name: milvus port: 19530 - nodePort: 32530 protocol: TCP targetPort: milvus - name: metrics protocol: TCP port: 9091 - nodePort: 32531 targetPort: metrics selector: app.kubernetes.io/name: milvus diff --git a/deploy/milvus-slim/milvus.yaml b/deploy/milvus-slim/milvus.yaml index fe68a8e..34adb68 100644 --- a/deploy/milvus-slim/milvus.yaml +++ b/deploy/milvus-slim/milvus.yaml @@ -101,8 +101,9 @@ spec: - emptyDir: {} name: tools - name: milvus-config - configMap: - name: openserverless-milvus + secret: + secretName: openserverless-milvus + defaultMode: 0440 - name: milvus-data-disk persistentVolumeClaim: claimName: openserverless-milvus diff --git a/deploy/milvus/milvus-cfg-base.yaml b/deploy/milvus/milvus-cfg-base.yaml index 2e54635..ab2c91e 100644 --- a/deploy/milvus/milvus-cfg-base.yaml +++ b/deploy/milvus/milvus-cfg-base.yaml @@ -17,11 +17,12 @@ # --- apiVersion: v1 -kind: ConfigMap +kind: Secret metadata: name: openserverless-milvus namespace: openserverless -data: +type: Opaque +stringData: default.yaml: |+ # Copyright (C) 2019-2021 Zilliz. All rights reserved. # @@ -38,7 +39,7 @@ data: rootPath: milvus auth: enabled: true - userName: root + userName: etcdmilvus password: 0therPa55 endpoints: - openserverless-etcd-0.openserverless-etcd-headless.openserverless.svc.cluster.local:2379 diff --git a/deploy/milvus/milvus-svc.yaml b/deploy/milvus/milvus-svc.yaml index 65143f6..637fa0e 100644 --- a/deploy/milvus/milvus-svc.yaml +++ b/deploy/milvus/milvus-svc.yaml @@ -69,17 +69,15 @@ metadata: app.kubernetes.io/instance: openserverless-milvus component: "standalone" spec: - type: NodePort + type: ClusterIP ports: - name: milvus port: 19530 - nodePort: 32530 protocol: TCP targetPort: milvus - name: metrics protocol: TCP port: 9091 - nodePort: 32531 targetPort: metrics selector: app.kubernetes.io/name: milvus diff --git a/deploy/milvus/milvus.yaml b/deploy/milvus/milvus.yaml index fe68a8e..34adb68 100644 --- a/deploy/milvus/milvus.yaml +++ b/deploy/milvus/milvus.yaml @@ -101,8 +101,9 @@ spec: - emptyDir: {} name: tools - name: milvus-config - configMap: - name: openserverless-milvus + secret: + secretName: openserverless-milvus + defaultMode: 0440 - name: milvus-data-disk persistentVolumeClaim: claimName: openserverless-milvus diff --git a/deploy/openserverless-permissions/operator-roles.yaml b/deploy/openserverless-permissions/operator-roles.yaml index 770d224..7df96fe 100644 --- a/deploy/openserverless-permissions/operator-roles.yaml +++ b/deploy/openserverless-permissions/operator-roles.yaml @@ -101,6 +101,11 @@ rules: resources: ["milvus","milvus/status","milvus/finalizers"] verbs: ["get","patch","list","update","watch","create","delete"] +# Datastore ingress isolation is reconciled by the operator. +- apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] + --- kind: RoleBinding apiVersion: rbac.authorization.k8s.io/v1 diff --git a/openserverless/config.py b/openserverless/config.py index 8815171..e89eab6 100644 --- a/openserverless/config.py +++ b/openserverless/config.py @@ -259,4 +259,4 @@ def detect(): def dump_config(): import openserverless.config as cfg for k in cfg.getall(): - logging.debug(f"{k} = {cfg.get(k)}") + logging.debug("configuration key present: %s", k) diff --git a/openserverless/datastore_security.py b/openserverless/datastore_security.py new file mode 100644 index 0000000..12925a0 --- /dev/null +++ b/openserverless/datastore_security.py @@ -0,0 +1,108 @@ +# 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. + +"""Private application of datastore credentials and narrow ingress policies.""" +import base64 +import json +import subprocess + +import openserverless.kube as kube + + +def apply_private(spec, namespace='openserverless'): + # The generic wrapper includes stdin and stderr in error logs. + result = subprocess.run(['kubectl', '-n', namespace, 'apply', '-f', '-'], + input=json.dumps(spec), text=True, capture_output=True, + timeout=60) + if result.returncode: + raise RuntimeError('Could not apply private datastore resources') + + +def apply(spec): + items = spec.get('items', [spec]) + private = [item for item in items if item['kind'] == 'Secret'] + public = [item for item in items if item['kind'] != 'Secret'] + if private: + apply_private({'apiVersion': 'v1', 'kind': 'List', 'items': private}) + return kube.apply({'apiVersion': 'v1', 'kind': 'List', 'items': public}) + + +def ensure_jwt_key(owner=None): + """Keep a per-installation key; never reuse the legacy repository key.""" + name = 'openserverless-etcd-jwt-token-v2' + raw = kube.kubectl('get', 'secret', name, '--ignore-not-found', '-o', 'json', + debugresult=False) + if raw.strip(): + key = base64.b64decode(json.loads(raw)['data']['jwt-token.pem'], validate=True) + check = subprocess.run(['openssl', 'rsa', '-check', '-noout'], input=key, + capture_output=True, timeout=30) + if check.returncode: + raise RuntimeError('Invalid etcd signing key; refusing implicit rotation') + return name + # A missing key on an already patched cluster is a security state error. + current = kube.kubectl('get', 'sts', 'openserverless-etcd', '--ignore-not-found', + '-o', 'json', debugresult=False) + if current.strip(): + volumes = json.loads(current)['spec']['template']['spec'].get('volumes', []) + if any(v.get('secret', {}).get('secretName') == name for v in volumes): + raise RuntimeError('Missing etcd signing key; refusing implicit rotation') + result = subprocess.run(['openssl', 'genpkey', '-algorithm', 'RSA', + '-pkeyopt', 'rsa_keygen_bits:3072'], + capture_output=True, timeout=30) + if result.returncode: + raise RuntimeError('Could not generate etcd signing key') + resource = {'apiVersion': 'v1', 'kind': 'Secret', 'type': 'Opaque', + 'metadata': {'name': name, 'namespace': 'openserverless'}, + 'data': {'jwt-token.pem': base64.b64encode(result.stdout).decode()}} + if owner: + import kopf + kopf.append_owner_reference([resource], owner) + apply_private(resource) + return name + + +def policy(component): + if component == 'etcd': + selector = {'app.kubernetes.io/name': 'openserverless-etcd'} + ingress = [ + {'from': [{'podSelector': {'matchLabels': selector}}], + 'ports': [{'port': 2379, 'protocol': 'TCP'}, {'port': 2380, 'protocol': 'TCP'}]}, + {'from': [{'podSelector': {'matchLabels': {'name': 'openserverless-operator'}}}, + {'podSelector': {'matchLabels': {'name': 'openserverless-milvus'}}}], + 'ports': [{'port': 2379, 'protocol': 'TCP'}]}] + elif component == 'milvus': + selector = {'name': 'openserverless-milvus'} + ingress = [ + {'ports': [{'port': 19530, 'protocol': 'TCP'}]}, + {'from': [{'podSelector': {'matchLabels': { + 'app.kubernetes.io/name': 'prometheus', + 'app.kubernetes.io/instance': 'openserverless-prometheus'}}}, + {'podSelector': {'matchLabels': {'name': 'openserverless-operator'}}}], + 'ports': [{'port': 9091, 'protocol': 'TCP'}]}] + else: + raise ValueError('Unknown datastore component') + return {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy', + 'metadata': {'name': 'openserverless-'+component+'-isolation', 'namespace': 'openserverless'}, + 'spec': {'podSelector': {'matchLabels': selector}, + 'policyTypes': ['Ingress'], 'ingress': ingress}} + + +def finish_milvus_reconcile(): + kube.kubectl('rollout', 'status', 'deployment/openserverless-milvus-standalone', '--timeout=300s') + # Remove the old ConfigMap AND its last-applied annotation by deleting it. + # Its contents include S3 and Milvus admin credentials as well as etcd root. + kube.kubectl('delete', 'configmap', 'openserverless-milvus', '--ignore-not-found') diff --git a/openserverless/etcd.py b/openserverless/etcd.py index 6d14f1f..b9a9536 100644 --- a/openserverless/etcd.py +++ b/openserverless/etcd.py @@ -21,6 +21,8 @@ import openserverless.config as cfg import openserverless.template as ntp import openserverless.util as util +from openserverless import datastore_security +from openserverless.etcd_admin import create_etcd_user, delete_db_user import os.path import logging import kopf @@ -44,15 +46,20 @@ def create(owner=None): kust = kus.patchTemplates("etcd",tplp , data) kust += kus.patchGenericEntry("Secret","openserverless-etcd-secret","/data/rootPassword",util.b64_encode(data['root_password'])) - spec = kus.kustom_list("etcd", kust, templates=spec_templates, data=data) + datastore_security.ensure_jwt_key(owner) + spec = kus.restricted_kustom_list("etcd", kust, templates=spec_templates, data=data, + templates_filter=["etcd-secret.yaml", "etcd-sts.yaml", "etcd-svc.yaml"]) + spec["items"].append(datastore_security.policy("etcd")) if owner: kopf.append_owner_reference(spec['items'], owner) else: cfg.put("state.etcd.spec", spec) - res = kube.apply(spec) + res = datastore_security.apply(spec) + kube.kubectl("rollout", "status", "statefulset/openserverless-etcd", "--timeout=300s") wait_for_etcd_ready() + kube.kubectl("delete", "secret", "openserverless-etcd-jwt-token", "--ignore-not-found") logging.info(f"create etcd: {res}") return res @@ -61,78 +68,6 @@ def wait_for_etcd_ready(): # dynamically detect etcd pod and wait for readiness util.wait_for_pod_ready(r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/name == 'openserverless-etcd')].metadata.name}") -def render_etcd_script(namespace,template,data): - """ - uses the given template to render a sh script to execute via bash shell. - """ - out = f"/tmp/__{namespace}_{template}" - file = ntp.spool_template(template, out, data) - return os.path.abspath(file) - -def exec_etcd_script(pod_name,path_to_etcd_script): - logging.info(f"passing script {path_to_etcd_script} to pod {pod_name}") - res = kube.kubectl("cp",path_to_etcd_script,f"{pod_name}:{path_to_etcd_script}") - res = kube.kubectl("exec","-i",pod_name,"--","/bin/bash","-c",f"chmod +x {path_to_etcd_script}") - res = kube.kubectl("exec","-i",pod_name,"--","/bin/bash","-c",path_to_etcd_script) - os.remove(path_to_etcd_script) - return res - - -def create_etcd_user(username:str, password:str, prefix:str): - """ - Creates a new ETCD username with the given password and assign redwrite permission on the given prefix - return: True if operation succeed, False otherwise. - """ - logging.info(f"authorizing new ETCD user {username} on prefix {prefix}") - - try: - data = {} - data["username"]=username - data["password"]=password - data["prefix"]=prefix - data["mode"]="create" - - path_to_etcd_script = render_etcd_script(username,"etcd_manage_user_tpl.sh",data) - pod_name = util.get_pod_name_by_selector("name=openserverless-etcd","{.items[0].metadata.name}") - - if(pod_name): - res = exec_etcd_script(pod_name,path_to_etcd_script) - if res: - return True - else: - logging.error(f"failed to add ETCD username {username}") - return False - - return False - except Exception as e: - logging.error(f"failed to add ETCD username {username}: {e}") - return False - -def delete_db_user(username): - """ - Reomves the specified user from the ETCD instance. - """ - logging.info(f"removing ETCD user {username}") - try: - data = {} - data["username"]=username - data["mode"]="delete" - - path_to_etcd_script = render_etcd_script(username,"etcd_manager_user_tpl.sh",data) - pod_name = util.get_pod_name_by_selector("name=openserverless-etcd","{.items[0].metadata.name}") - - if(pod_name): - res = exec_etcd_script(pod_name,path_to_etcd_script) - if res: - return res - else: - logging.error(f"failed to remove ETCD username {username}") - - return None - except Exception as e: - logging.error(f"failed to remove ETCD username {username} authorization id and key: {e}") - return None - def delete_by_owner(): spec = kus.build("etcd") res = kube.delete(spec) diff --git a/openserverless/etcd_admin.py b/openserverless/etcd_admin.py new file mode 100644 index 0000000..73b7f6c --- /dev/null +++ b/openserverless/etcd_admin.py @@ -0,0 +1,113 @@ +# 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. + +"""Authenticated etcd 3.5 administration without shell interpolation.""" +import base64 +import contextlib +import logging +import re +import requests +import openserverless.util as util + +def _post(session, endpoint, path, body, allowed=()): + for attempt in range(2): + try: + r=session.post(endpoint+'/v3/'+path,json=body,timeout=8) + result=r.json() + except (requests.RequestException,ValueError): + raise RuntimeError('etcd administration request failed') from None + error=result.get('error') or result.get('message','') + credentials=getattr(session,'_ops_etcd_credentials',None) + if attempt==0 and credentials and error in ( + 'etcdserver: revision of auth store is old', 'etcdserver: invalid auth token'): + # Auth-store updates can invalidate a JWT between admin operations. + # Refresh only this authenticated session, never substitute root for a tenant. + try: + auth=session.post(endpoint+'/v3/auth/authenticate',json=credentials,timeout=8) + token=auth.json().get('token') if auth.ok else None + except (requests.RequestException,ValueError): token=None + if not token: raise RuntimeError('Could not renew etcd administration authentication') + session.headers['Authorization']=token + continue + if error in allowed: return result + if not r.ok or error: + raise RuntimeError('etcd administration request was rejected') + return result + + +@contextlib.contextmanager +def _admin(): + data = util.get_etcd_config_data() + endpoint = 'http://' + data['name'] + ':2379' + with requests.Session() as session: + session.trust_env = False + session._ops_etcd_credentials = {'name': 'root', 'password': data['root_password']} + session.headers['Authorization'] = _post(session, endpoint, 'auth/authenticate', + session._ops_etcd_credentials)['token'] + yield session, endpoint + + +def create_etcd_user(username, password, prefix, readonly_keys=()): + if not re.fullmatch(r'[A-Za-z0-9_-]{1,64}',username) or username=='root' or not prefix.strip('/'): + raise ValueError('Invalid etcd user or key prefix') + try: + with _admin() as (session,endpoint): + result=_post(session,endpoint,'auth/user/add',{'name':username,'password':password}, + allowed=('etcdserver: user name already exists',)) + if result.get('error') or result.get('message'): + _post(session,endpoint,'auth/authenticate',{'name':username,'password':password}) + role=username+'_role' + _post(session,endpoint,'auth/role/add',{'name':role}, + allowed=('etcdserver: role name already exists',)) + key=(prefix.rstrip('/')+'/').encode() + end=bytearray(key);end[-1]+=1 + _post(session,endpoint,'auth/role/grant',{'name':role,'perm':{'permType':'READWRITE', + 'key':base64.b64encode(key).decode(),'range_end':base64.b64encode(end).decode()}}) + for exact_key in readonly_keys: + _post(session, endpoint, 'auth/role/grant', {'name': role, 'perm': { + 'permType': 'READ', 'key': base64.b64encode(exact_key.encode()).decode()}}) + _post(session,endpoint,'auth/user/grant',{'user':username,'role':role}) + identity = _post(session, endpoint, 'auth/user/get', {'name': username}) + permissions = _post(session, endpoint, 'auth/role/get', {'role': role}).get('perm', []) + expected = {'permType': 'READWRITE', 'key': base64.b64encode(key).decode(), + 'range_end': base64.b64encode(end).decode()} + expected_permissions = [expected] + [ + {'permType': 'READ', 'key': base64.b64encode(k.encode()).decode()} + for k in readonly_keys] + def normalized(values): + return sorted((v.get('permType', 'READ'), v['key'], v.get('range_end', '')) for v in values) + if identity.get('roles') != [role] or normalized(permissions) != normalized(expected_permissions): + raise RuntimeError('Existing etcd account exceeds the required prefix scope') + return True + except RuntimeError: + logging.error('Could not provision the etcd user') + return False + + +def delete_db_user(username): + if not re.fullmatch(r'[A-Za-z0-9_-]{1,64}',username) or username=='root': + raise ValueError('Invalid etcd user') + try: + with _admin() as (session,endpoint): + _post(session,endpoint,'auth/user/delete',{'name':username}, + allowed=('etcdserver: user name not found',)) + _post(session,endpoint,'auth/role/delete',{'role':username+'_role'}, + allowed=('etcdserver: role name not found',)) + return True + except RuntimeError: + logging.error('Could not remove the etcd user') + return False diff --git a/openserverless/milvus_standalone.py b/openserverless/milvus_standalone.py index dfa15a2..b48ba87 100644 --- a/openserverless/milvus_standalone.py +++ b/openserverless/milvus_standalone.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. # +from openserverless import etcd, datastore_security import time import kopf, logging @@ -95,12 +96,14 @@ def create(owner=None): kust = patchEntries(data) mspec = kus.kustom_list(dir, kust, templates=[], data=data) + mspec["items"].append(datastore_security.policy("milvus")) if owner: kopf.append_owner_reference(mspec['items'], owner) else: cfg.put("state.milvus.spec", mspec) - kube.apply(mspec) + datastore_security.apply(mspec) + datastore_security.finish_milvus_reconcile() util.wait_for_pod_ready( r"{.items[?(@.metadata.labels.app\.kubernetes\.io\/instance == 'openserverless-milvus')].metadata.name}") @@ -133,9 +136,9 @@ def create_milvus_accounts(data: dict): """" Creates technical accounts for ETCD and SEAWEEDFS """ - # currently we use the ETCD root password, so we skip the ETCD user creation. - # res = util.check(etcd.create_etcd_user(data['milvus_etcd_username'],data['milvus_etcd_password'],data['milvus_etcd_prefix']),"create_etcd_milvus_user",True) - + if not etcd.create_etcd_user(data['milvus_etcd_username'], data['milvus_etcd_password'], + data['milvus_etcd_prefix'], readonly_keys=('health',)): + return False return create_seaweedfs_milvus_account(data) def delete_seaweedfs_milvus_account(data: dict): @@ -156,7 +159,8 @@ def delete_milvus_accounts(data: dict): """" Deletes technical accounts for ETCD and SEAWEEDFS """ - # currently we use the ETCD root password, so we skip the ETCD user deletion. + if not etcd.delete_db_user(data["milvus_etcd_username"]): + return False logging.info("removing milvus technical accounts.") return delete_seaweedfs_milvus_account(data) @@ -168,6 +172,25 @@ def create_default_milvus_database(data): """ logging.info("*** configuring MILVUS database for openserverless") adminClient = MilvusAdminClient() + # Security reconciliation must preserve an already provisioned system user. + # Authenticate with its configured credential; never reset it on an error. + from openserverless.milvus_simple_client import MilvusSimpleClient + import requests + try: + client = MilvusSimpleClient(uri=adminClient.milvus_url, token=adminClient.milvus_admin_token) + users = client.list_users() + if "openserverless" in users: + response = requests.post(adminClient.milvus_url + '/v2/vectordb/collections/list', + headers={'Authorization': 'Bearer openserverless:' + data['openserverless_password']}, + json={'dbName': 'openserverless'}, timeout=10) + result = response.json() + if not response.ok or result.get('code') != 0 or 'data' not in result: + return False + _annotate_nuv_milvus_metadata(data) + return True + except (requests.RequestException, ValueError, TypeError): + logging.error('Could not verify the existing Milvus system account') + return False res = adminClient.setup_user("openserverless", data["openserverless_password"], "openserverless") if (res): diff --git a/openserverless/templates/milvus-cfg-base.yaml b/openserverless/templates/milvus-cfg-base.yaml index 38f8035..ac1d7c6 100644 --- a/openserverless/templates/milvus-cfg-base.yaml +++ b/openserverless/templates/milvus-cfg-base.yaml @@ -17,11 +17,12 @@ # --- apiVersion: v1 -kind: ConfigMap +kind: Secret metadata: name: openserverless-milvus namespace: openserverless -data: +type: Opaque +stringData: default.yaml: |+ # Copyright (C) 2019-2021 Zilliz. All rights reserved. # @@ -38,8 +39,8 @@ data: rootPath: milvus auth: enabled: true - userName: root - password: {{milvus_etcd_root_password}} + userName: {{milvus_etcd_username | tojson}} + password: {{milvus_etcd_password | tojson}} endpoints: {% for idx in etcd_range -%} - {{etcd_container}}-{{idx}}.openserverless-etcd-headless.openserverless.svc.cluster.local:2379 @@ -52,8 +53,8 @@ data: minio: address: {{bucket_server_hostname}} port: {{bucket_server_port}} - accessKeyID: {{milvus_s3_username}} - secretAccessKey: {{milvus_s3_password}} + accessKeyID: {{milvus_s3_username | tojson}} + secretAccessKey: {{milvus_s3_password | tojson}} useSSL: false bucketName: vectors rootPath: files @@ -140,4 +141,4 @@ data: common: security: authorizationEnabled: true - defaultRootPassword: {{milvus_root_password}} \ No newline at end of file + defaultRootPassword: {{milvus_root_password | tojson}} \ No newline at end of file diff --git a/openserverless/templates/milvus-cfg-slim-base.yaml b/openserverless/templates/milvus-cfg-slim-base.yaml index 0ede540..d834180 100644 --- a/openserverless/templates/milvus-cfg-slim-base.yaml +++ b/openserverless/templates/milvus-cfg-slim-base.yaml @@ -17,11 +17,12 @@ # --- apiVersion: v1 -kind: ConfigMap +kind: Secret metadata: name: openserverless-milvus namespace: openserverless -data: +type: Opaque +stringData: default.yaml: |+ # Copyright (C) 2019-2021 Zilliz. All rights reserved. # @@ -38,8 +39,8 @@ data: rootPath: milvus auth: enabled: true - userName: root - password: {{milvus_etcd_root_password}} + userName: {{milvus_etcd_username | tojson}} + password: {{milvus_etcd_password | tojson}} endpoints: {% for idx in etcd_range -%} - {{etcd_container}}-{{idx}}.openserverless-etcd-headless.openserverless.svc.cluster.local:2379 @@ -51,8 +52,8 @@ data: minio: address: {{bucket_server_hostname}} port: {{bucket_server_port}} - accessKeyID: {{milvus_s3_username}} - secretAccessKey: {{milvus_s3_password}} + accessKeyID: {{milvus_s3_username | tojson}} + secretAccessKey: {{milvus_s3_password | tojson}} useSSL: false bucketName: vectors rootPath: files @@ -129,4 +130,4 @@ data: common: security: authorizationEnabled: true - defaultRootPassword: {{milvus_root_password}} \ No newline at end of file + defaultRootPassword: {{milvus_root_password | tojson}} \ No newline at end of file diff --git a/tests/test_datastore_security.py b/tests/test_datastore_security.py new file mode 100644 index 0000000..5b0701d --- /dev/null +++ b/tests/test_datastore_security.py @@ -0,0 +1,178 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import base64 +import json +import logging +from pathlib import Path +import subprocess +import unittest +from unittest.mock import patch + +from openserverless import config, datastore_security as security + + +class SecurityTests(unittest.TestCase): + def test_debug_logs_never_contain_values(self): + config.clean() + config.configure({'etcd': {'root': {'password': 'sentinel-password'}}, + 'opaque': {'url': 'http://user:sentinel-uri@internal'}, + 'state': {'spec': {'secret': 'sentinel-nested'}}}) + with self.assertLogs(level=logging.DEBUG) as logs: + config.dump_config() + text = '\n'.join(logs.output) + for value in ['sentinel-password', 'sentinel-uri', 'sentinel-nested']: + self.assertNotIn(value, text) + self.assertIn('etcd.root.password', text) + + def test_private_apply_does_not_report_stderr_or_payload(self): + spec = {'kind': 'Secret', 'stringData': {'password': 'sentinel'}} + with patch.object(security.subprocess, 'run', return_value= + subprocess.CompletedProcess([], 1, 'sentinel', 'sentinel')): + with self.assertRaisesRegex(RuntimeError, '^Could not apply private datastore resources$'): + security.apply_private(spec) + + def test_secrets_do_not_reach_public_kubectl_wrapper(self): + secret = {'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'name': 'private'}} + service = {'apiVersion': 'v1', 'kind': 'Service', 'metadata': {'name': 'public'}} + with patch.object(security, 'apply_private') as private, patch.object(security.kube, 'apply') as public: + security.apply({'items': [secret, service]}) + self.assertEqual(private.call_args.args[0]['items'], [secret]) + self.assertEqual(public.call_args.args[0]['items'], [service]) + + def test_fresh_signing_keys_are_unique_and_private(self): + captured = [] + with patch.object(security.kube, 'kubectl', return_value=''), \ + patch.object(security, 'apply_private', side_effect=captured.append): + security.ensure_jwt_key() + security.ensure_jwt_key() + keys = [base64.b64decode(x['data']['jwt-token.pem']) for x in captured] + self.assertNotEqual(keys[0], keys[1]) + self.assertTrue(all(b'PRIVATE KEY' in key for key in keys)) + + def test_existing_key_is_preserved(self): + key = subprocess.run(['openssl', 'genpkey', '-algorithm', 'RSA', + '-pkeyopt', 'rsa_keygen_bits:2048'], capture_output=True, check=True).stdout + raw = json.dumps({'data': {'jwt-token.pem': base64.b64encode(key).decode()}}) + with patch.object(security.kube, 'kubectl', return_value=raw), \ + patch.object(security, 'apply_private') as apply: + security.ensure_jwt_key() + apply.assert_not_called() + + def test_invalid_key_is_not_replaced(self): + raw = json.dumps({'data': {'jwt-token.pem': base64.b64encode(b'not-a-key').decode()}}) + with patch.object(security.kube, 'kubectl', return_value=raw), \ + patch.object(security, 'apply_private') as apply: + with self.assertRaises(RuntimeError): + security.ensure_jwt_key() + apply.assert_not_called() + + def test_missing_key_on_patched_cluster_is_not_reset(self): + sts = {'spec': {'template': {'spec': {'volumes': [ + {'secret': {'secretName': 'openserverless-etcd-jwt-token-v2'}}]}}}} + with patch.object(security.kube, 'kubectl', side_effect=['', json.dumps(sts)]), \ + patch.object(security, 'apply_private') as apply: + with self.assertRaises(RuntimeError): + security.ensure_jwt_key() + apply.assert_not_called() + + def test_monitoring_does_not_allow_all_peers(self): + rules = security.policy('milvus')['spec']['ingress'] + metrics = [rule for rule in rules if any(p['port'] == 9091 for p in rule['ports'])] + self.assertEqual(len(metrics), 1) + self.assertTrue(all(peer['podSelector']['matchLabels'] for peer in metrics[0]['from'])) + self.assertFalse(any('namespaceSelector' in peer for peer in metrics[0]['from'])) + + def test_etcd_peer_port_is_restricted_to_members(self): + rules = security.policy('etcd')['spec']['ingress'] + peer_rule = [rule for rule in rules if any(p['port'] == 2380 for p in rule['ports'])] + self.assertEqual(len(peer_rule), 1) + self.assertEqual(peer_rule[0]['from'], [{'podSelector': {'matchLabels': { + 'app.kubernetes.io/name': 'openserverless-etcd'}}}]) + + def test_secret_templates_cover_full_and_slim(self): + import jinja2 + import jinja2.meta + import yaml + root = Path(__file__).resolve().parents[1] + for mode in ['base', 'slim-base']: + path = root/'openserverless/templates'/('milvus-cfg-'+mode+'.yaml') + source = path.read_text() + variables = jinja2.meta.find_undeclared_variables(jinja2.Environment().parse(source)) + data = {name: 'sentinel' for name in variables} + data.update(etcd_range=[0, 1, 2], milvus_etcd_username='etcdmilvus', + milvus_etcd_password='punctuation:"\\\n', milvus_max_database_num=64, + milvus_max_role_num=128, milvus_max_user_num=128) + obj = yaml.safe_load(jinja2.Template(source).render(**data)) + self.assertEqual(obj['kind'], 'Secret') + cfg = yaml.safe_load(obj['stringData']['default.yaml']) + self.assertEqual(cfg['etcd']['auth']['userName'], 'etcdmilvus') + self.assertEqual(cfg['etcd']['auth']['password'], data['milvus_etcd_password']) + self.assertNotEqual(cfg['etcd']['auth']['userName'], 'root') + + def test_existing_system_user_is_verified_without_recreation(self): + from openserverless import milvus_standalone as milvus + data = {'openserverless_password': 'synthetic'} + with patch.object(milvus, 'MilvusAdminClient') as admin, \ + patch('openserverless.milvus_simple_client.MilvusSimpleClient') as client, \ + patch('requests.post') as post, patch.object(milvus, '_annotate_nuv_milvus_metadata'): + client.return_value.list_users.return_value = ['root', 'openserverless'] + post.return_value.ok = True + post.return_value.json.return_value = {'code': 0, 'data': []} + self.assertTrue(milvus.create_default_milvus_database(data)) + admin.return_value.setup_user.assert_not_called() + post.return_value.json.return_value = {'code': 1800, 'message': 'denied'} + self.assertFalse(milvus.create_default_milvus_database(data)) + admin.return_value.setup_user.assert_not_called() + + def test_scoped_account_accepts_only_its_prefix_and_exact_health_read(self): + import contextlib + try: + from openserverless import etcd_admin as admin + except ImportError: + from openserverless import etcd as admin + def operation(session, endpoint, path, body, **kwargs): + if path == 'auth/user/get': return {'roles': ['probe_role']} + if path == 'auth/role/get': return {'perm': [ + {'key': base64.b64encode(b'health').decode()}, + {'permType': 'READWRITE', 'key': base64.b64encode(b'milvus/').decode(), + 'range_end': base64.b64encode(b'milvus0').decode()}]} + return {} + with patch.object(admin, '_admin', return_value=contextlib.nullcontext((object(), 'endpoint'))), \ + patch.object(admin, '_post', side_effect=operation): + self.assertTrue(admin.create_etcd_user('probe', 'quoted:\"\n', 'milvus', readonly_keys=('health',))) + # The existing health permission is not silently accepted for a generic user. + self.assertFalse(admin.create_etcd_user('probe', 'quoted', 'milvus')) + + def test_reconciliation_refuses_an_existing_root_role(self): + import contextlib + try: + from openserverless import etcd_admin as admin + except ImportError: + from openserverless import etcd as admin + def operation(session, endpoint, path, body, **kwargs): + if path == 'auth/user/get': return {'roles': ['root', 'probe_role']} + if path == 'auth/role/get': return {'perm': []} + return {} + with patch.object(admin, '_admin', return_value=contextlib.nullcontext((object(), 'endpoint'))), \ + patch.object(admin, '_post', side_effect=operation): + self.assertFalse(admin.create_etcd_user('probe', 'synthetic', 'milvus')) + + +if __name__ == '__main__': + import jinja2.meta + unittest.main() From bd5b15f0b70491ea306f9d6d86c2f7af3d7994b7 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 19 Oct 2026 14:33:41 +0200 Subject: [PATCH 2/3] Run the shared security regression suite in PR checks --- .github/workflows/check.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 205f79e..ec74171 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -64,8 +64,8 @@ jobs: run: bash .github/cisetup.sh - name: Install run: task setup - - name: Datastore security regression tests - run: poetry run python3 -m unittest discover -s tests -p test_datastore_security.py -v + - name: Security regression tests + run: poetry run python3 -m unittest discover -s tests -p 'test_*.py' -v - name: BuildAndLoad run: task build-and-load - name: AddIngress From f06f3ec866f52ebfe5cff45deb5e4256438d1c0e Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 19 Oct 2026 15:08:28 +0200 Subject: [PATCH 3/3] Use private tunnels for external etcd and Milvus integration tests --- .github/workflows/check.yml | 19 ++++ openserverless/datastore_connection.py | 69 ++++++++++++++ openserverless/etcd_admin.py | 5 +- tests/kind/milvus_standalone_slim_test.ipy | 22 ++++- tests/kind/milvus_standalone_test.ipy | 22 ++++- tests/test_datastore_connection.py | 100 +++++++++++++++++++++ 6 files changed, 231 insertions(+), 6 deletions(-) create mode 100644 openserverless/datastore_connection.py create mode 100644 tests/test_datastore_connection.py diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index ec74171..965ca64 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -76,3 +76,22 @@ jobs: - name: Integration Test run: task itest continue-on-error: false + - name: Reject new integration failures + run: | + # The legacy runner writes failures without returning nonzero. + # These two tests also fail on unmodified 0.9.0 (run 38893750250). + unexpected_failures=0 + if [ -f _failed.txt ]; then + while IFS= read -r failed_test; do + case "$failed_test" in + tests/kind/ferretdb_test.ipy|tests/kind/runtimes_preloader_test.ipy) + echo "::warning::Pre-existing 0.9.0 integration failure: $failed_test" + ;; + *) + echo "::error::Integration regression: $failed_test" + unexpected_failures=1 + ;; + esac + done < _failed.txt + fi + exit "$unexpected_failures" diff --git a/openserverless/datastore_connection.py b/openserverless/datastore_connection.py new file mode 100644 index 0000000..5fd21c8 --- /dev/null +++ b/openserverless/datastore_connection.py @@ -0,0 +1,69 @@ +# 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. +# + +"""Reach a private service in-cluster or through an authenticated local tunnel.""" +import contextlib +import os +import re +import selectors +import subprocess +import time + + +@contextlib.contextmanager +def service_endpoint(service, port, namespace='openserverless'): + if not re.fullmatch(r'[a-z0-9][a-z0-9.-]*', service) or not re.fullmatch(r'[a-z0-9][a-z0-9-]*', namespace): + raise ValueError('Invalid datastore service') + if not isinstance(port, int) or not 0 < port < 65536: + raise ValueError('Invalid datastore port') + if os.environ.get('KUBERNETES_SERVICE_HOST'): + yield service, port + return + + # Use the current kubeconfig and namespace, as with the operator's other + # kubectl calls. Never expose a public listener or put credentials in argv. + process = subprocess.Popen( + ['kubectl', '-n', namespace, 'port-forward', '--address=127.0.0.1', + 'service/' + service, ':' + str(port)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + with selectors.DefaultSelector() as selector: + selector.register(process.stdout, selectors.EVENT_READ) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if selector.select(timeout=min(1, max(0, deadline-time.monotonic()))): + line = process.stdout.readline() + match = re.fullmatch(r'Forwarding from 127\.0\.0\.1:(\d+) -> ' + str(port), line.strip()) + if match: + local_port = int(match.group(1)) + if not 0 < local_port < 65536: + break + yield '127.0.0.1', local_port + return + if process.poll() is not None: + break + raise RuntimeError('Could not open the private datastore tunnel') + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + process.stdout.close() diff --git a/openserverless/etcd_admin.py b/openserverless/etcd_admin.py index 73b7f6c..fe8ed52 100644 --- a/openserverless/etcd_admin.py +++ b/openserverless/etcd_admin.py @@ -22,6 +22,7 @@ import re import requests import openserverless.util as util +from openserverless.datastore_connection import service_endpoint def _post(session, endpoint, path, body, allowed=()): for attempt in range(2): @@ -52,8 +53,8 @@ def _post(session, endpoint, path, body, allowed=()): @contextlib.contextmanager def _admin(): data = util.get_etcd_config_data() - endpoint = 'http://' + data['name'] + ':2379' - with requests.Session() as session: + with service_endpoint(data['name'], 2379) as (host, port), requests.Session() as session: + endpoint = 'http://' + host + ':' + str(port) session.trust_env = False session._ops_etcd_credentials = {'name': 'root', 'password': data['root_password']} session.headers['Authorization'] = _post(session, endpoint, 'auth/authenticate', diff --git a/tests/kind/milvus_standalone_slim_test.ipy b/tests/kind/milvus_standalone_slim_test.ipy index a8374c2..cf0c297 100644 --- a/tests/kind/milvus_standalone_slim_test.ipy +++ b/tests/kind/milvus_standalone_slim_test.ipy @@ -18,6 +18,10 @@ # import os +import atexit +import contextlib +from unittest.mock import patch +from openserverless.datastore_connection import service_endpoint import openserverless.config as cfg import openserverless.milvus_standalone as milvus @@ -55,7 +59,20 @@ assert(etcd.create()) assert(seaweedfs.create()) -assert(milvus.create()) +endpoint_stack = contextlib.ExitStack() +atexit.register(endpoint_stack.close) +finish_reconcile = milvus.datastore_security.finish_milvus_reconcile +local_endpoint = None + +def finish_and_connect(): + global local_endpoint + finish_reconcile() + local_endpoint = endpoint_stack.enter_context(service_endpoint('openserverless-milvus', 19530)) + os.environ['MILVUS_API_HOST'] = local_endpoint[0] + os.environ['MILVUS_API_PORT'] = str(local_endpoint[1]) + +with patch.object(milvus.datastore_security, 'finish_milvus_reconcile', side_effect=finish_and_connect): + assert(milvus.create()) # generate some user config ucfg = UserConfig({ @@ -93,7 +110,7 @@ assert port # test client -client = MilvusClient(uri=f'http://localhost:{port}',token=token, db_name=database) +client = MilvusClient(uri=f'http://{local_endpoint[0]}:{local_endpoint[1]}',token=token, db_name=database) client.create_collection("testcollection", dimension=128) collections = client.list_collections() or [] assert("testcollection" in collections) @@ -105,6 +122,7 @@ assert(milvus.delete_ow_milvus(ucfg)) assert(milvus.delete()) assert(etcd.delete()) assert(seaweedfs.delete()) +endpoint_stack.close() # final cleanup tu.run_proc("kubectl -n openserverless delete all --all") diff --git a/tests/kind/milvus_standalone_test.ipy b/tests/kind/milvus_standalone_test.ipy index ec45216..2931c0f 100644 --- a/tests/kind/milvus_standalone_test.ipy +++ b/tests/kind/milvus_standalone_test.ipy @@ -18,6 +18,10 @@ # import os +import atexit +import contextlib +from unittest.mock import patch +from openserverless.datastore_connection import service_endpoint import openserverless.config as cfg import openserverless.etcd as etcd @@ -52,7 +56,20 @@ os.environ['MILVUS_API_HOST']='localhost' assert(etcd.create()) assert(seaweedfs.create()) -assert(milvus.create()) +endpoint_stack = contextlib.ExitStack() +atexit.register(endpoint_stack.close) +finish_reconcile = milvus.datastore_security.finish_milvus_reconcile +local_endpoint = None + +def finish_and_connect(): + global local_endpoint + finish_reconcile() + local_endpoint = endpoint_stack.enter_context(service_endpoint('openserverless-milvus', 19530)) + os.environ['MILVUS_API_HOST'] = local_endpoint[0] + os.environ['MILVUS_API_PORT'] = str(local_endpoint[1]) + +with patch.object(milvus.datastore_security, 'finish_milvus_reconcile', side_effect=finish_and_connect): + assert(milvus.create()) # generate some user config ucfg = UserConfig({ @@ -89,7 +106,7 @@ assert database assert port # test client -client = MilvusClient(uri=f'http://localhost:{port}',token=token, db_name=database) +client = MilvusClient(uri=f'http://{local_endpoint[0]}:{local_endpoint[1]}',token=token, db_name=database) client.create_collection("testcollection", dimension=128) collections = client.list_collections() or [] assert("testcollection" in collections) @@ -102,6 +119,7 @@ assert(milvus.delete_ow_milvus(ucfg)) assert(milvus.delete()) assert(etcd.delete()) assert(seaweedfs.delete()) +endpoint_stack.close() tu.run_proc("kubectl -n openserverless delete all --all") tu.run_proc("kubectl -n openserverless delete pvc --all") diff --git a/tests/test_datastore_connection.py b/tests/test_datastore_connection.py new file mode 100644 index 0000000..dd0fea9 --- /dev/null +++ b/tests/test_datastore_connection.py @@ -0,0 +1,100 @@ +# 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 io +import os +import subprocess +import unittest +from unittest.mock import MagicMock, patch + +from openserverless import datastore_connection as connection + + +class PrivateServiceConnectionTests(unittest.TestCase): + def process(self, line='Forwarding from 127.0.0.1:41237 -> 2379\n'): + process = MagicMock() + process.stdout = io.StringIO(line) + process.poll.return_value = None + return process + + def test_in_cluster_uses_service_without_starting_a_tunnel(self): + with patch.dict(os.environ, {'KUBERNETES_SERVICE_HOST': 'cluster'}, clear=True), \ + patch.object(connection.subprocess, 'Popen') as spawn: + with connection.service_endpoint('openserverless-etcd', 2379) as endpoint: + self.assertEqual(endpoint, ('openserverless-etcd', 2379)) + spawn.assert_not_called() + def test_external_tunnel_binds_only_loopback_and_closes(self): + process = self.process() + with patch.dict(os.environ, {}, clear=True), \ + patch.object(connection.subprocess, 'Popen', return_value=process) as spawn, \ + patch.object(connection.selectors, 'DefaultSelector') as selector: + selector.return_value.__enter__.return_value.select.return_value = [object()] + with connection.service_endpoint('openserverless-etcd', 2379) as endpoint: + self.assertEqual(endpoint, ('127.0.0.1', 41237)) + command = spawn.call_args.args[0] + self.assertIn('--address=127.0.0.1', command) + self.assertIn('service/openserverless-etcd', command) + self.assertIn(':2379', command) + self.assertFalse(any('password' in item for item in command)) + process.terminate.assert_called_once() + self.assertTrue(process.stdout.closed) + + def test_consumer_error_still_closes_tunnel(self): + process = self.process() + with patch.dict(os.environ, {}, clear=True), \ + patch.object(connection.subprocess, 'Popen', return_value=process), \ + patch.object(connection.selectors, 'DefaultSelector') as selector: + selector.return_value.__enter__.return_value.select.return_value = [object()] + with self.assertRaisesRegex(ValueError, 'consumer failed'): + with connection.service_endpoint('openserverless-etcd', 2379): + raise ValueError('consumer failed') + process.terminate.assert_called_once() + self.assertTrue(process.stdout.closed) + + def test_failed_tunnel_does_not_report_process_output(self): + process = self.process('credential-bearing diagnostic must stay private\n') + process.poll.return_value = 1 + with patch.dict(os.environ, {}, clear=True), \ + patch.object(connection.subprocess, 'Popen', return_value=process), \ + patch.object(connection.selectors, 'DefaultSelector') as selector: + selector.return_value.__enter__.return_value.select.return_value = [object()] + with self.assertRaisesRegex(RuntimeError, '^Could not open the private datastore tunnel$'): + with connection.service_endpoint('openserverless-etcd', 2379): + self.fail('A failed tunnel must not yield an endpoint') + + def test_a_stuck_tunnel_is_killed_during_cleanup(self): + process = self.process() + process.wait.side_effect = [subprocess.TimeoutExpired('kubectl', 5), None] + with patch.dict(os.environ, {}, clear=True), \ + patch.object(connection.subprocess, 'Popen', return_value=process), \ + patch.object(connection.selectors, 'DefaultSelector') as selector: + selector.return_value.__enter__.return_value.select.return_value = [object()] + with connection.service_endpoint('openserverless-etcd', 2379): + pass + process.kill.assert_called_once() + self.assertTrue(process.stdout.closed) + + def test_invalid_targets_cannot_be_kubectl_options(self): + with patch.object(connection.subprocess, 'Popen') as spawn: + for service, port in [('--all-namespaces', 2379), ('service;command', 2379), + ('redis', 0), ('redis', 65536)]: + with self.assertRaises(ValueError): + with connection.service_endpoint(service, port): + self.fail('Invalid target accepted') + spawn.assert_not_called()