e0e362d700
SDK Tests / changes (push) Successful in 2m29s
Real E2E Tests / changes (push) Successful in 2m29s
Deploy Docs Pages / build (push) Has been cancelled
Deploy Docs Pages / deploy (push) Has been cancelled
Real E2E Tests / JavaScript E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Python E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Java E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / C# E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Go E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Real E2E CI (push) Has been cancelled
SDK Tests / SDK CI (push) Has been cancelled
SDK Tests / CLI Tests (push) Has been cancelled
SDK Tests / Python SDK Quality (code-interpreter) (push) Has been cancelled
SDK Tests / Python SDK Quality (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (sandbox) (push) Has been cancelled
SDK Tests / CLI Quality (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Go SDK Quality And Tests (push) Has been cancelled
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
# Copyright 2026 Alibaba Group Holding Ltd.
|
|
#
|
|
# Licensed 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.
|
|
|
|
"""Kubernetes V1SecurityContext ↔ plain dict helpers for CRD pod specs."""
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
def build_security_context_from_dict(
|
|
security_context_dict: Dict[str, Any],
|
|
) -> Optional[Any]:
|
|
"""
|
|
Convert a security context dict to ``V1SecurityContext``.
|
|
|
|
Empty dict returns None.
|
|
"""
|
|
if not security_context_dict:
|
|
return None
|
|
|
|
from kubernetes.client import V1SecurityContext, V1Capabilities
|
|
|
|
capabilities = None
|
|
if "capabilities" in security_context_dict:
|
|
caps_dict = security_context_dict["capabilities"]
|
|
add_caps = caps_dict.get("add", [])
|
|
drop_caps = caps_dict.get("drop", [])
|
|
capabilities = V1Capabilities(
|
|
add=add_caps if add_caps else None,
|
|
drop=drop_caps if drop_caps else None,
|
|
)
|
|
|
|
privileged = security_context_dict.get("privileged")
|
|
|
|
if capabilities is None and privileged is None:
|
|
return None
|
|
|
|
return V1SecurityContext(
|
|
capabilities=capabilities,
|
|
privileged=privileged,
|
|
)
|
|
|
|
|
|
def serialize_security_context_to_dict(
|
|
security_context: Optional[Any],
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Serialize ``V1SecurityContext`` to a CRD-friendly dict."""
|
|
if not security_context:
|
|
return None
|
|
|
|
result: Dict[str, Any] = {}
|
|
|
|
if security_context.capabilities:
|
|
caps: Dict[str, Any] = {}
|
|
if security_context.capabilities.add:
|
|
caps["add"] = security_context.capabilities.add
|
|
if security_context.capabilities.drop:
|
|
caps["drop"] = security_context.capabilities.drop
|
|
if caps:
|
|
result["capabilities"] = caps
|
|
|
|
if security_context.privileged is not None:
|
|
result["privileged"] = security_context.privileged
|
|
|
|
if getattr(security_context, "seccomp_profile", None) is not None:
|
|
sp = security_context.seccomp_profile
|
|
profile_dict: Dict[str, Any] = {"type": sp.type}
|
|
if getattr(sp, "localhost_profile", None) is not None:
|
|
profile_dict["localhostProfile"] = sp.localhost_profile
|
|
result["seccompProfile"] = profile_dict
|
|
|
|
if getattr(security_context, "app_armor_profile", None) is not None:
|
|
ap = security_context.app_armor_profile
|
|
profile_dict = {"type": ap.type}
|
|
if getattr(ap, "localhost_profile", None) is not None:
|
|
profile_dict["localhostProfile"] = ap.localhost_profile
|
|
result["appArmorProfile"] = profile_dict
|
|
|
|
return result if result else None
|