24621395421

Committed 19 Apr 2026 04:58AM UTC coverage: 81.992% (+4.1%) from 77.939%

Build # 24621395421

Build Type

push

github

Committed by

web-flow

Commit Message

Add OpenAPI v3 validation with CEL rules; extract validator into k8s_schema (#104)

## Summary

Extracts all OpenAPI validation out of `k8s.py` / `k8s_api.py` into a
new `kubernator/plugins/k8s_schema/` package, and adds an OpenAPI v3
validator with client-side CEL rule evaluation as the default on
clusters ≥ 1.27.

- **`k8s_api.py` no longer does validation** — `K8SResourcePluginMixin`
holds a single `self.validator` and delegates every manifest check
(minimal envelope → rdef lookup → full schema → CEL rules) to it.
Dependency graph is one-way: `k8s_schema → k8s_api`.
- **Two validators behind a factory.** `SwaggerV2Validator` preserves
current behaviour (swagger.json + OAS31). `OpenAPIV3Validator` is new:
cluster-first `/openapi/v3` discovery with GitHub fallback, lazy
per-group fetch, transitive `$ref` closure across group documents.
- **K8s extensions enforced client-side** under v3:
`x-kubernetes-list-type` (`atomic` / `set` / `map` with
`list-map-keys`), `-preserve-unknown-fields`, `-embedded-resource`,
`-int-or-string`.
- **CEL rules evaluated client-side.** Lazy-compiled, cached by rule
text for the run. Ports of the K8s CEL libraries — `lists`, `regex`,
`format.named(...)`, `quantity`, `IP`, `CIDR` — plus a compatibility
shim for `optional.of` / `optional.none` / `hasValue` / `value` /
`orValue` so `optionalSelf` / `optionalOldSelf` work. Transition rules
fire in the apply path against the cluster's current state.
- **Selection.** `ktor.k8s.register()` gains `openapi_version` (`auto` /
`v2` / `v3`) and `openapi_source` (`auto` / `cluster` / `github`) kwargs
(also readable from context). `auto` gates on server minor ≥ 1.27 and
falls back to v2 on any v3 failure. `istio.py` is migrated to the same
factory.

New dep: `cel-python ~=0.5`.

## Test plan

- [x] 111 unit tests (v2 parity; v3 lazy fetch; api_versions without
sub-doc fetch; source fallback; every extension keyword; CEL walker +
evaluator; 7 extension libraries; factory selection / vers... (continued)

Coverage Stats

943 of 1328 branches covered (71.01%)

Branch coverage included in aggregate %.

1040 of 1082 new or added lines in 18 files covered. (96.12%)

6 existing lines in 2 files now uncovered.

3979 of 4675 relevant lines covered (85.11%)

4.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.24

/src/main/python/kubernator/plugins/k8s_schema/base.py

# -*- coding: utf-8 -*-
#
#   Copyright 2020 Express Systems USA, Inc
#   Copyright 2026 Karellen, Inc.
#
#   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.
#

from __future__ import annotations

import base64
from collections.abc import Iterable, Iterator, Mapping, MutableMapping
from typing import Literal, Optional

from jsonschema._format import FormatChecker
from jsonschema.exceptions import ValidationError
from jsonschema.validators import Draft7Validator

from kubernator.plugins.k8s_api import (K8SResourceDef,
                                        K8SResourceDefKey,
                                        to_group_and_version)


K8S_MINIMAL_RESOURCE_SCHEMA = {
    "properties": {
        "apiVersion": {"type": "string"},
        "kind": {"type": "string"},
        "metadata": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "namespace": {"type": "string"},
            },
            "required": ["name"],
        },
    },
    "type": "object",
    "required": ["apiVersion", "kind"],
}
K8S_MINIMAL_RESOURCE_VALIDATOR = Draft7Validator(K8S_MINIMAL_RESOURCE_SCHEMA)


def extract_gvk_keys(obj: Mapping) -> Iterable[K8SResourceDefKey]:
    """Yield :class:`K8SResourceDefKey` for each ``x-kubernetes-group-version-kind``
    entry on *obj*. Handles both the single-dict and list-of-dicts shapes
    that K8s OpenAPI documents use."""
    gvks = obj.get("x-kubernetes-group-version-kind")
    if not gvks:
        return
    if isinstance(gvks, Mapping):
        gvks = [gvks]
    for gvk in gvks:
        yield K8SResourceDefKey(gvk["group"], gvk["version"], gvk["kind"])


def is_integer(instance):
    # bool inherits from int, so ensure bools aren't reported as ints
    if isinstance(instance, bool):
        return False
    return isinstance(instance, int)


def is_string(instance):
    return isinstance(instance, str)


def type_validator(validator, data_type, instance, schema):
    if instance is None:
        return

    if (data_type == "string"
            and (schema.get("format") == "int-or-string"
                 or schema.get("x-kubernetes-int-or-string") is True)):
        if not (is_string(instance) or is_integer(instance)):
            yield ValidationError("%r is not of type %s" % (instance, "int-or-string"))
    elif not validator.is_type(instance, data_type):
        yield ValidationError("%r is not of type %s" % (instance, data_type))


k8s_format_checker = FormatChecker()


@k8s_format_checker.checks("int32")
def check_int32(value):
    return value is not None and (-2147483648 < value < 2147483647)


@k8s_format_checker.checks("int64")
def check_int64(value):
    return value is not None and (-9223372036854775808 < value < 9223372036854775807)


@k8s_format_checker.checks("float")
def check_float(value):
    return value is not None and (-3.4E+38 < value < +3.4E+38)


@k8s_format_checker.checks("double")
def check_double(value):
    return value is not None and (-1.7E+308 < value < +1.7E+308)


@k8s_format_checker.checks("byte", ValueError)
def check_byte(value):
    if value is None:
        return False
    base64.b64decode(value, validate=True)
    return True


@k8s_format_checker.checks("int-or-string")
def check_int_or_string(value):
    return check_int32(value) if is_integer(value) else is_string(value)


class OpenAPIValidator:
    """Concrete base class for OpenAPI-backed Kubernetes manifest validators.

    Subclasses implement :meth:`load`, :meth:`iter_errors`, and
    :meth:`api_versions`; this base supplies the manifest-level
    entry points (:meth:`iter_manifest_errors`, :meth:`get_manifest_rdef`)
    shared by v2 and v3 so the K8s plugin mixin can delegate all
    validation to the validator without caring about version or
    minimal-schema details.
    """

    version: Literal["v2", "v3"]
    resource_definitions: MutableMapping[K8SResourceDefKey, K8SResourceDef]
    resource_paths: MutableMapping[K8SResourceDefKey, Mapping[str, dict]]

    def load(self) -> None:
        raise NotImplementedError

    def iter_errors(
            self,
            manifest: Mapping,
            rdef: K8SResourceDef,
            *,
            old_manifest: Optional[Mapping] = None,
    ) -> Iterator[ValidationError]:
        raise NotImplementedError

    def api_versions(self) -> Iterable[str]:
        raise NotImplementedError

    # -- manifest-level validation shared across versions ----------------

    def get_manifest_rdef(self, manifest: Mapping) -> K8SResourceDef:
        """Resolve a manifest to its :class:`K8SResourceDef`.
        Raises :class:`ValidationError` if the kind is unknown."""
        key = K8SResourceDefKey(*to_group_and_version(manifest["apiVersion"]),
                                manifest["kind"])
        try:
            return self.resource_definitions[key]
        except KeyError:
            raise ValidationError(
                f"{key} is not a defined Kubernetes resource",
                validator=K8S_MINIMAL_RESOURCE_VALIDATOR,
                validator_value=key,
                instance=manifest,
                schema=K8S_MINIMAL_RESOURCE_SCHEMA,
            )

    def iter_manifest_errors(
            self,
            manifest: Mapping,
            *,
            old_manifest: Optional[Mapping] = None,
    ) -> Iterator[ValidationError]:
        """Validate a manifest end-to-end: minimal envelope check →
        rdef lookup → full schema (and CEL) validation."""
        sentinel: Optional[ValidationError] = None
        for err in K8S_MINIMAL_RESOURCE_VALIDATOR.iter_errors(manifest):
            sentinel = err
            yield err
        if sentinel is not None:
            return
        try:
            rdef = self.get_manifest_rdef(manifest)
        except ValidationError as e:
            yield e
            return
        yield from self.iter_errors(manifest, rdef, old_manifest=old_manifest)

1	# -- coding: utf-8 --
2	#
3	# Copyright 2020 Express Systems USA, Inc
4	# Copyright 2026 Karellen, Inc.
5	#
6	# Licensed under the Apache License, Version 2.0 (the "License");
7	# you may not use this file except in compliance with the License.
8	# You may obtain a copy of the License at
9	#
10	# http://www.apache.org/licenses/LICENSE-2.0
11	#
12	# Unless required by applicable law or agreed to in writing, software
13	# distributed under the License is distributed on an "AS IS" BASIS,
14	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	# See the License for the specific language governing permissions and
16	# limitations under the License.
17	#
18
19	from __future__ import annotations	5✔
20
21	import base64	5✔
22	from collections.abc import Iterable, Iterator, Mapping, MutableMapping	5✔
23	from typing import Literal, Optional	5✔
24
25	from jsonschema._format import FormatChecker	5✔
26	from jsonschema.exceptions import ValidationError	5✔
27	from jsonschema.validators import Draft7Validator	5✔
28
29	from kubernator.plugins.k8s_api import (K8SResourceDef,	5✔
30	K8SResourceDefKey,
31	to_group_and_version)
32
33
34	K8S_MINIMAL_RESOURCE_SCHEMA = {	5✔
35	"properties": {
36	"apiVersion": {"type": "string"},
37	"kind": {"type": "string"},
38	"metadata": {
39	"type": "object",
40	"properties": {
41	"name": {"type": "string"},
42	"namespace": {"type": "string"},
43	},
44	"required": ["name"],
45	},
46	},
47	"type": "object",
48	"required": ["apiVersion", "kind"],
49	}
50	K8S_MINIMAL_RESOURCE_VALIDATOR = Draft7Validator(K8S_MINIMAL_RESOURCE_SCHEMA)	5✔
51
52
53	def extract_gvk_keys(obj: Mapping) -> Iterable[K8SResourceDefKey]:	5✔
54	"""Yield :class:`K8SResourceDefKey` for each ``x-kubernetes-group-version-kind``
55	entry on obj. Handles both the single-dict and list-of-dicts shapes
56	that K8s OpenAPI documents use."""
57	gvks = obj.get("x-kubernetes-group-version-kind")	5✔
58	if not gvks:	5✔
59	return	5✔
60	if isinstance(gvks, Mapping):	5✔
61	gvks = [gvks]	5✔
62	for gvk in gvks:	5✔
63	yield K8SResourceDefKey(gvk["group"], gvk["version"], gvk["kind"])	5✔
64
65
66	def is_integer(instance):	5✔
67	# bool inherits from int, so ensure bools aren't reported as ints
68	if isinstance(instance, bool):	5✔
69	return False	5✔
70	return isinstance(instance, int)	5✔
71
72
73	def is_string(instance):	5✔
74	return isinstance(instance, str)	5✔
75
76
77	def type_validator(validator, data_type, instance, schema):	5✔
78	if instance is None:	5✔
79	return	5✔
80
81	if (data_type == "string"	5✔
82	and (schema.get("format") == "int-or-string"
83	or schema.get("x-kubernetes-int-or-string") is True)):
84	if not (is_string(instance) or is_integer(instance)):	5✔
85	yield ValidationError("%r is not of type %s" % (instance, "int-or-string"))	5✔
86	elif not validator.is_type(instance, data_type):	5✔
87	yield ValidationError("%r is not of type %s" % (instance, data_type))	5✔
88
89
90	k8s_format_checker = FormatChecker()	5✔
91
92
93	@k8s_format_checker.checks("int32")	5✔
94	def check_int32(value):	5✔
95	return value is not None and (-2147483648 < value < 2147483647)	5✔
96
97
98	@k8s_format_checker.checks("int64")	5✔
99	def check_int64(value):	5✔
100	return value is not None and (-9223372036854775808 < value < 9223372036854775807)	5✔
101
102
103	@k8s_format_checker.checks("float")	5✔
104	def check_float(value):	5✔
105	return value is not None and (-3.4E+38 < value < +3.4E+38)	5✔
106
107
108	@k8s_format_checker.checks("double")	5✔
109	def check_double(value):	5✔
110	return value is not None and (-1.7E+308 < value < +1.7E+308)	5✔
111
112
113	@k8s_format_checker.checks("byte", ValueError)	5✔
114	def check_byte(value):	5✔
115	if value is None:	5!
NEW 116	return False	×
117	base64.b64decode(value, validate=True)	5✔
118	return True	5✔
119
120
121	@k8s_format_checker.checks("int-or-string")	5✔
122	def check_int_or_string(value):	5✔
123	return check_int32(value) if is_integer(value) else is_string(value)	5✔
124
125
126	class OpenAPIValidator:	5✔
127	"""Concrete base class for OpenAPI-backed Kubernetes manifest validators.
128
129	Subclasses implement :meth:`load`, :meth:`iter_errors`, and
130	:meth:`api_versions`; this base supplies the manifest-level
131	entry points (:meth:`iter_manifest_errors`, :meth:`get_manifest_rdef`)
132	shared by v2 and v3 so the K8s plugin mixin can delegate all
133	validation to the validator without caring about version or
134	minimal-schema details.
135	"""
136
137	version: Literal["v2", "v3"]	5✔
138	resource_definitions: MutableMapping[K8SResourceDefKey, K8SResourceDef]	5✔
139	resource_paths: MutableMapping[K8SResourceDefKey, Mapping[str, dict]]	5✔
140
141	def load(self) -> None:	5✔
NEW 142	raise NotImplementedError	×
143
144	def iter_errors(	5✔
145	self,
146	manifest: Mapping,
147	rdef: K8SResourceDef,
148	*,
149	old_manifest: Optional[Mapping] = None,
150	) -> Iterator[ValidationError]:
NEW 151	raise NotImplementedError	×
152
153	def api_versions(self) -> Iterable[str]:	5✔
NEW 154	raise NotImplementedError	×
155
156	# -- manifest-level validation shared across versions ----------------
157
158	def get_manifest_rdef(self, manifest: Mapping) -> K8SResourceDef:	5✔
159	"""Resolve a manifest to its :class:`K8SResourceDef`.
160	Raises :class:`ValidationError` if the kind is unknown."""
161	key = K8SResourceDefKey(*to_group_and_version(manifest["apiVersion"]),	5✔
162	manifest["kind"])
163	try:	5✔
164	return self.resource_definitions[key]	5✔
165	except KeyError:	5✔
166	raise ValidationError(	5✔
167	f"{key} is not a defined Kubernetes resource",
168	validator=K8S_MINIMAL_RESOURCE_VALIDATOR,
169	validator_value=key,
170	instance=manifest,
171	schema=K8S_MINIMAL_RESOURCE_SCHEMA,
172	)
173
174	def iter_manifest_errors(	5✔
175	self,
176	manifest: Mapping,
177	*,
178	old_manifest: Optional[Mapping] = None,
179	) -> Iterator[ValidationError]:
180	"""Validate a manifest end-to-end: minimal envelope check →
181	rdef lookup → full schema (and CEL) validation."""
182	sentinel: Optional[ValidationError] = None	5✔
183	for err in K8S_MINIMAL_RESOURCE_VALIDATOR.iter_errors(manifest):	5✔
184	sentinel = err	5✔
185	yield err	5✔
186	if sentinel is not None:	5✔
187	return	5✔
188	try:	5✔
189	rdef = self.get_manifest_rdef(manifest)	5✔
190	except ValidationError as e:	5✔
191	yield e	5✔
192	return	5✔
193	yield from self.iter_errors(manifest, rdef, old_manifest=old_manifest)	5✔

karellen / kubernator / 24621395421

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous