11004195341

Committed 23 Sep 2024 11:41PM UTC coverage: 94.846% (-0.1%) from 94.973%

Build # 11004195341

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Support referencing TwingateGroup from TwingateResourceAccess (#412)

Run Details

258 of 286 branches covered (90.21%)

Branch coverage included in aggregate %.

28 of 30 new or added lines in 3 files covered. (93.33%)

1 existing line in 1 file now uncovered.

975 of 1014 relevant lines covered (96.15%)

0.96 hits per line

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

97.71

/app/crds.py

import logging
from collections.abc import MutableMapping
from datetime import datetime
from enum import Enum
from typing import Any

import kubernetes.client
import pendulum
from croniter import croniter
from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    ValidationInfo,
    field_validator,
    model_validator,
)
from pydantic.alias_generators import to_camel
from semantic_version import NpmSpec

from app.settings import get_settings
from app.version_policy_providers import get_provider

K8sObject = MutableMapping[Any, Any]
OptionalK8sObject = K8sObject | None


class K8sMetadata(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, extra="allow", alias_generator=to_camel
    )

    uid: str
    name: str
    namespace: str

    @property
    def owner_reference_object(self) -> dict:
        return {
            "apiVersion": "twingate.com/v1",
            "kind": "TwingateResource",
            "name": self.name,
            "uid": self.uid,
        }


class BaseK8sModel(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, extra="allow", alias_generator=to_camel
    )

    api_version: str
    kind: str
    metadata: K8sMetadata
    status: dict[str, Any] | None = None


class _KubernetesObjectRef(BaseModel):
    name: str
    namespace: str = Field(default="default")


# region TwingateResourceCRD


class ProtocolPolicy(str, Enum):
    ALLOW_ALL = "ALLOW_ALL"
    RESTRICTED = "RESTRICTED"


class ProtocolRange(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel
    )

    start: int = Field(ge=1, le=65535)
    end: int = Field(ge=1, le=65535)

    @model_validator(mode="after")
    def check_ports(self):
        if self.start > self.end:
            raise ValueError("Start port value must be less or equal to end port value")

        return self


class ResourceProtocol(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel
    )

    policy: ProtocolPolicy = ProtocolPolicy.ALLOW_ALL
    ports: list[ProtocolRange] = Field(default_factory=list)

    @model_validator(mode="after")
    def check_policy_ports(self):
        if self.policy == ProtocolPolicy.ALLOW_ALL and self.ports:
            raise ValueError("ports can't be set if policy is ALLOW_ALL")

        return self


class ResourceProtocols(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel
    )

    allow_icmp: bool | None = True
    tcp: ResourceProtocol = Field(default_factory=ResourceProtocol)
    udp: ResourceProtocol = Field(default_factory=ResourceProtocol)


class ResourceSpec(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel
    )

    id: str | None = None
    name: str
    address: str
    alias: str | None = None
    remote_network_id: str = Field(
        default_factory=lambda: get_settings().remote_network_id
    )
    security_policy_id: str | None = None
    is_visible: bool = True
    is_browser_shortcut_enabled: bool = True
    protocols: ResourceProtocols = Field(default_factory=ResourceProtocols)

    def __is_wildcard(self):
        return "*" in self.address or "?" in self.address

    @model_validator(mode="after")
    def check_address_if_is_browser_shortcut_enabled(self):
        if self.is_browser_shortcut_enabled and self.__is_wildcard():
            raise ValueError(
                "isBrowserShortcutEnabled can't be True for wildcard addresses"
            )

        return self


class TwingateResourceCRD(BaseK8sModel):
    model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")

    spec: ResourceSpec


# endregion

# region TwingateResourceAccessCRD


class PrincipalTypeEnum(str, Enum):
    Group = "group"
    ServiceAccount = "serviceAccount"


class _PrincipalExternalRef(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    type: PrincipalTypeEnum
    name: str


class ResourceAccessSpec(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    resource_ref: _KubernetesObjectRef
    principal_id: str | None = None
    group_ref: _KubernetesObjectRef | None = None
    principal_external_ref: _PrincipalExternalRef | None = None
    security_policy_id: str | None = None

    @model_validator(mode="after")
    def validate_target_ref_exists(self):
        if self.principal_id or self.group_ref or self.principal_external_ref:
            return self

        raise ValueError("Missing principal_id, group_ref or principal_external_ref")

    @property
    def resource_ref_fullname(self) -> str:
        return f"{self.resource_ref.namespace}/{self.resource_ref.name}"

    @property
    def group_ref_fullname(self) -> str | None:
        return (
            f"{self.group_ref.namespace}/{self.group_ref.name}"
            if self.group_ref
            else None
        )

    def _get_ref_object(
        self, plural_type: str, namespace: str, name: str
    ) -> OptionalK8sObject:
        log_prefix = (
            f"ResourceAccessCRD._get_ref_object({plural_type}, {namespace}, {name}):"
        )
        try:
            kapi = kubernetes.client.CustomObjectsApi()
            response = kapi.get_namespaced_custom_object(
                "twingate.com",
                "v1beta",
                namespace,
                plural_type,
                name,
            )
            logging.info(
                "%s got %s",
                log_prefix,
                response,
            )
            return response
        except kubernetes.client.exceptions.ApiException as api_ex:
            if api_ex.status == 404:
                logging.warning("%s resource not found.", log_prefix)
            else:
                logging.exception("%s failed", log_prefix)

            return None

    def get_resource_ref_object(self) -> OptionalK8sObject:
        return self._get_ref_object(
            "twingateresources", self.resource_ref.namespace, self.resource_ref.name
        )

    def get_group_ref_object(self) -> OptionalK8sObject:
        if not self.group_ref:
            return None

        return self._get_ref_object(
            "twingategroups", self.group_ref.namespace, self.group_ref.name
        )

    def get_resource(self) -> TwingateResourceCRD | None:
        resource_ref_object = self.get_resource_ref_object()
        if not resource_ref_object:
            return None

        resource = ResourceSpec(**resource_ref_object.pop("spec"))
        metadata = K8sMetadata(**resource_ref_object.pop("metadata"))
        return TwingateResourceCRD(
            metadata=metadata, spec=resource, **resource_ref_object
        )


class TwingateResourceAccessCRD(BaseK8sModel):
    model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")

    spec: ResourceAccessSpec


# endregion

# region TwingateConnector


class ConnectorImagePolicyProvidersEnum(str, Enum):
    dockerhub = "dockerhub"
    google = "google"


class ConnectorImagePolicy(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
    )

    provider: ConnectorImagePolicyProvidersEnum = (
        ConnectorImagePolicyProvidersEnum.dockerhub
    )
    repository: str = "twingate/connector"
    schedule: str = "0 0 * * *"
    version: str = "^1.0.0"
    allow_prerelease: bool = False

    @field_validator("version")
    @classmethod
    def check_valid_version_specifier(cls, v: str, info: ValidationInfo) -> str:
        try:
            NpmSpec(v)
        except ValueError as vex:
            raise ValueError("Invalid version specifier") from vex
        return v

    @field_validator("schedule")
    @classmethod
    def check_valid_crontab(cls, v: str, info: ValidationInfo) -> str:
        if v:
            try:
                croniter(v)
            except ValueError as vex:
                raise ValueError("Invalid schedule value") from vex
        return v

    def get_next_date_iso8601(self) -> str:
        next_date = croniter(self.schedule, pendulum.now("UTC")).get_next(datetime)
        return pendulum.instance(next_date).to_iso8601_string()

    def get_image(self) -> str:
        provider = get_provider(self.provider.value, repository=self.repository)
        if tag := provider.get_latest(
            self.version, allow_prerelease=self.allow_prerelease
        ):
            return f"{self.repository}:{tag}"

        raise ValueError(
            f"Could not find valid tag for '{self.version}' at '{self.repository}'"
        )


class ConnectorImage(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
    )

    repository: str = "twingate/connector"
    tag: str = "1"

    def __str__(self):
        return f"{self.repository}:{self.tag}"


class ConnectorSpec(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
    )

    id: str | None = None
    name: str | None = None
    log_level: int = 3
    log_analytics: bool = True
    has_status_notifications_enabled: bool = True
    image: ConnectorImage | None = None
    image_policy: ConnectorImagePolicy | None = None
    container_extra: dict[str, Any] = {}
    pod_extra: dict[str, Any] = {}
    pod_annotations: dict[str, Any] = {}
    sidecar_containers: list[dict[str, Any]] = []

    remote_network_id: str = Field(
        default_factory=lambda: get_settings().remote_network_id
    )

    @model_validator(mode="after")
    def validate_image_or_image_policy(self):
        if self.image or self.image_policy:
            return self

        # Default to having `image`
        return self.model_copy(update=dict(image=ConnectorImage()))

    def get_image(self) -> str:
        if image := self.image:
            return str(image)

        if image_policy := self.image_policy:
            return image_policy.get_image()

        # Impossible to get here because of our model validator
        raise ValueError("Invalid ConnectorSpec")  # pragma: no cover


class TwingateConnectorCRD(BaseK8sModel):
    model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")

    spec: ConnectorSpec


# endregion

# region TwingateGroup


class GroupSpec(BaseModel):
    model_config = ConfigDict(
        frozen=True, populate_by_name=True, alias_generator=to_camel
    )

    id: str | None = None
    name: str


class TwingateGroupCRD(BaseK8sModel):
    model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")

    spec: GroupSpec


# endregion

1	import logging	1✔
2	from collections.abc import MutableMapping	1✔
3	from datetime import datetime	1✔
4	from enum import Enum	1✔
5	from typing import Any	1✔
6
7	import kubernetes.client	1✔
8	import pendulum	1✔
9	from croniter import croniter	1✔
10	from pydantic import (	1✔
11	BaseModel,
12	ConfigDict,
13	Field,
14	ValidationInfo,
15	field_validator,
16	model_validator,
17	)
18	from pydantic.alias_generators import to_camel	1✔
19	from semantic_version import NpmSpec	1✔
20
21	from app.settings import get_settings	1✔
22	from app.version_policy_providers import get_provider	1✔
23
24	K8sObject = MutableMapping[Any, Any]	1✔
25	OptionalK8sObject = K8sObject \| None	1✔
26
27
28	class K8sMetadata(BaseModel):	1✔
29	model_config = ConfigDict(	1✔
30	frozen=True, populate_by_name=True, extra="allow", alias_generator=to_camel
31	)
32
33	uid: str	1✔
34	name: str	1✔
35	namespace: str	1✔
36
37	@property	1✔
38	def owner_reference_object(self) -> dict:	1✔
39	return {	1✔
40	"apiVersion": "twingate.com/v1",
41	"kind": "TwingateResource",
42	"name": self.name,
43	"uid": self.uid,
44	}
45
46
47	class BaseK8sModel(BaseModel):	1✔
48	model_config = ConfigDict(	1✔
49	frozen=True, populate_by_name=True, extra="allow", alias_generator=to_camel
50	)
51
52	api_version: str	1✔
53	kind: str	1✔
54	metadata: K8sMetadata	1✔
55	status: dict[str, Any] \| None = None	1✔
56
57
58	class _KubernetesObjectRef(BaseModel):	1✔
59	name: str	1✔
60	namespace: str = Field(default="default")	1✔
61
62
63	# region TwingateResourceCRD
64
65
66	class ProtocolPolicy(str, Enum):	1✔
67	ALLOW_ALL = "ALLOW_ALL"	1✔
68	RESTRICTED = "RESTRICTED"	1✔
69
70
71	class ProtocolRange(BaseModel):	1✔
72	model_config = ConfigDict(	1✔
73	frozen=True, populate_by_name=True, alias_generator=to_camel
74	)
75
76	start: int = Field(ge=1, le=65535)	1✔
77	end: int = Field(ge=1, le=65535)	1✔
78
79	@model_validator(mode="after")	1✔
80	def check_ports(self):	1✔
81	if self.start > self.end:	1✔
82	raise ValueError("Start port value must be less or equal to end port value")	1✔
83
84	return self	1✔
85
86
87	class ResourceProtocol(BaseModel):	1✔
88	model_config = ConfigDict(	1✔
89	frozen=True, populate_by_name=True, alias_generator=to_camel
90	)
91
92	policy: ProtocolPolicy = ProtocolPolicy.ALLOW_ALL	1✔
93	ports: list[ProtocolRange] = Field(default_factory=list)	1✔
94
95	@model_validator(mode="after")	1✔
96	def check_policy_ports(self):	1✔
97	if self.policy == ProtocolPolicy.ALLOW_ALL and self.ports:	1✔
98	raise ValueError("ports can't be set if policy is ALLOW_ALL")	1✔
99
100	return self	1✔
101
102
103	class ResourceProtocols(BaseModel):	1✔
104	model_config = ConfigDict(	1✔
105	frozen=True, populate_by_name=True, alias_generator=to_camel
106	)
107
108	allow_icmp: bool \| None = True	1✔
109	tcp: ResourceProtocol = Field(default_factory=ResourceProtocol)	1✔
110	udp: ResourceProtocol = Field(default_factory=ResourceProtocol)	1✔
111
112
113	class ResourceSpec(BaseModel):	1✔
114	model_config = ConfigDict(	1✔
115	frozen=True, populate_by_name=True, alias_generator=to_camel
116	)
117
118	id: str \| None = None	1✔
119	name: str	1✔
120	address: str	1✔
121	alias: str \| None = None	1✔
122	remote_network_id: str = Field(	1✔
123	default_factory=lambda: get_settings().remote_network_id
124	)
125	security_policy_id: str \| None = None	1✔
126	is_visible: bool = True	1✔
127	is_browser_shortcut_enabled: bool = True	1✔
128	protocols: ResourceProtocols = Field(default_factory=ResourceProtocols)	1✔
129
130	def __is_wildcard(self):	1✔
131	return "*" in self.address or "?" in self.address	1✔
132
133	@model_validator(mode="after")	1✔
134	def check_address_if_is_browser_shortcut_enabled(self):	1✔
135	if self.is_browser_shortcut_enabled and self.__is_wildcard():	1✔
136	raise ValueError(	1✔
137	"isBrowserShortcutEnabled can't be True for wildcard addresses"
138	)
139
140	return self	1✔
141
142
143	class TwingateResourceCRD(BaseK8sModel):	1✔
144	model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")	1✔
145
146	spec: ResourceSpec	1✔
147
148
149	# endregion
150
151	# region TwingateResourceAccessCRD
152
153
154	class PrincipalTypeEnum(str, Enum):	1✔
155	Group = "group"	1✔
156	ServiceAccount = "serviceAccount"	1✔
157
158
159	class _PrincipalExternalRef(BaseModel):	1✔
160	model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)	1✔
161
162	type: PrincipalTypeEnum	1✔
163	name: str	1✔
164
165
166	class ResourceAccessSpec(BaseModel):	1✔
167	model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)	1✔
168
169	resource_ref: _KubernetesObjectRef	1✔
170	principal_id: str \| None = None	1✔
171	group_ref: _KubernetesObjectRef \| None = None	1✔
172	principal_external_ref: _PrincipalExternalRef \| None = None	1✔
173	security_policy_id: str \| None = None	1✔
174
175	@model_validator(mode="after")	1✔
176	def validate_target_ref_exists(self):	1✔
177	if self.principal_id or self.group_ref or self.principal_external_ref:	1!
178	return self	1✔
179
NEW 180	raise ValueError("Missing principal_id, group_ref or principal_external_ref")	×
181
182	@property	1✔
183	def resource_ref_fullname(self) -> str:	1✔
184	return f"{self.resource_ref.namespace}/{self.resource_ref.name}"	1✔
185
186	@property	1✔
187	def group_ref_fullname(self) -> str \| None:	1✔
NEW 188	return (	×
189	f"{self.group_ref.namespace}/{self.group_ref.name}"
190	if self.group_ref
191	else None
192	)
193
194	def _get_ref_object(	1✔
195	self, plural_type: str, namespace: str, name: str
196	) -> OptionalK8sObject:
197	log_prefix = (	1✔
198	f"ResourceAccessCRD._get_ref_object({plural_type}, {namespace}, {name}):"
199	)
200	try:	1✔
201	kapi = kubernetes.client.CustomObjectsApi()	1✔
202	response = kapi.get_namespaced_custom_object(	1✔
203	"twingate.com",
204	"v1beta",
205	namespace,
206	plural_type,
207	name,
208	)
209	logging.info(	1✔
210	"%s got %s",
211	log_prefix,
212	response,
213	)
214	return response	1✔
215	except kubernetes.client.exceptions.ApiException as api_ex:	1✔
216	if api_ex.status == 404:	1✔
217	logging.warning("%s resource not found.", log_prefix)	1✔
218	else:
219	logging.exception("%s failed", log_prefix)	1✔
220
221	return None	1✔
222
223	def get_resource_ref_object(self) -> OptionalK8sObject:	1✔
224	return self._get_ref_object(	1✔
225	"twingateresources", self.resource_ref.namespace, self.resource_ref.name
226	)
227
228	def get_group_ref_object(self) -> OptionalK8sObject:	1✔
229	if not self.group_ref:	1!
UNCOV 230	return None	×
231
232	return self._get_ref_object(	1✔
233	"twingategroups", self.group_ref.namespace, self.group_ref.name
234	)
235
236	def get_resource(self) -> TwingateResourceCRD \| None:	1✔
237	resource_ref_object = self.get_resource_ref_object()	1✔
238	if not resource_ref_object:	1✔
239	return None	1✔
240
241	resource = ResourceSpec(**resource_ref_object.pop("spec"))	1✔
242	metadata = K8sMetadata(**resource_ref_object.pop("metadata"))	1✔
243	return TwingateResourceCRD(	1✔
244	metadata=metadata, spec=resource, **resource_ref_object
245	)
246
247
248	class TwingateResourceAccessCRD(BaseK8sModel):	1✔
249	model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")	1✔
250
251	spec: ResourceAccessSpec	1✔
252
253
254	# endregion
255
256	# region TwingateConnector
257
258
259	class ConnectorImagePolicyProvidersEnum(str, Enum):	1✔
260	dockerhub = "dockerhub"	1✔
261	google = "google"	1✔
262
263
264	class ConnectorImagePolicy(BaseModel):	1✔
265	model_config = ConfigDict(	1✔
266	frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
267	)
268
269	provider: ConnectorImagePolicyProvidersEnum = (	1✔
270	ConnectorImagePolicyProvidersEnum.dockerhub
271	)
272	repository: str = "twingate/connector"	1✔
273	schedule: str = "0 0 * * *"	1✔
274	version: str = "^1.0.0"	1✔
275	allow_prerelease: bool = False	1✔
276
277	@field_validator("version")	1✔
278	@classmethod	1✔
279	def check_valid_version_specifier(cls, v: str, info: ValidationInfo) -> str:	1✔
280	try:	1✔
281	NpmSpec(v)	1✔
282	except ValueError as vex:	1✔
283	raise ValueError("Invalid version specifier") from vex	1✔
284	return v	1✔
285
286	@field_validator("schedule")	1✔
287	@classmethod	1✔
288	def check_valid_crontab(cls, v: str, info: ValidationInfo) -> str:	1✔
289	if v:	1!
290	try:	1✔
291	croniter(v)	1✔
292	except ValueError as vex:	1✔
293	raise ValueError("Invalid schedule value") from vex	1✔
294	return v	1✔
295
296	def get_next_date_iso8601(self) -> str:	1✔
297	next_date = croniter(self.schedule, pendulum.now("UTC")).get_next(datetime)	1✔
298	return pendulum.instance(next_date).to_iso8601_string()	1✔
299
300	def get_image(self) -> str:	1✔
301	provider = get_provider(self.provider.value, repository=self.repository)	1✔
302	if tag := provider.get_latest(	1✔
303	self.version, allow_prerelease=self.allow_prerelease
304	):
305	return f"{self.repository}:{tag}"	1✔
306
307	raise ValueError(	1✔
308	f"Could not find valid tag for '{self.version}' at '{self.repository}'"
309	)
310
311
312	class ConnectorImage(BaseModel):	1✔
313	model_config = ConfigDict(	1✔
314	frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
315	)
316
317	repository: str = "twingate/connector"	1✔
318	tag: str = "1"	1✔
319
320	def __str__(self):	1✔
321	return f"{self.repository}:{self.tag}"	1✔
322
323
324	class ConnectorSpec(BaseModel):	1✔
325	model_config = ConfigDict(	1✔
326	frozen=True, populate_by_name=True, alias_generator=to_camel, extra="allow"
327	)
328
329	id: str \| None = None	1✔
330	name: str \| None = None	1✔
331	log_level: int = 3	1✔
332	log_analytics: bool = True	1✔
333	has_status_notifications_enabled: bool = True	1✔
334	image: ConnectorImage \| None = None	1✔
335	image_policy: ConnectorImagePolicy \| None = None	1✔
336	container_extra: dict[str, Any] = {}	1✔
337	pod_extra: dict[str, Any] = {}	1✔
338	pod_annotations: dict[str, Any] = {}	1✔
339	sidecar_containers: list[dict[str, Any]] = []	1✔
340
341	remote_network_id: str = Field(	1✔
342	default_factory=lambda: get_settings().remote_network_id
343	)
344
345	@model_validator(mode="after")	1✔
346	def validate_image_or_image_policy(self):	1✔
347	if self.image or self.image_policy:	1✔
348	return self	1✔
349
350	# Default to having `image`
351	return self.model_copy(update=dict(image=ConnectorImage()))	1✔
352
353	def get_image(self) -> str:	1✔
354	if image := self.image:	1✔
355	return str(image)	1✔
356
357	if image_policy := self.image_policy:	1✔
358	return image_policy.get_image()	1✔
359
360	# Impossible to get here because of our model validator
361	raise ValueError("Invalid ConnectorSpec") # pragma: no cover
362
363
364	class TwingateConnectorCRD(BaseK8sModel):	1✔
365	model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")	1✔
366
367	spec: ConnectorSpec	1✔
368
369
370	# endregion
371
372	# region TwingateGroup
373
374
375	class GroupSpec(BaseModel):	1✔
376	model_config = ConfigDict(	1✔
377	frozen=True, populate_by_name=True, alias_generator=to_camel
378	)
379
380	id: str \| None = None	1✔
381	name: str	1✔
382
383
384	class TwingateGroupCRD(BaseK8sModel):	1✔
385	model_config = ConfigDict(frozen=True, populate_by_name=True, extra="allow")	1✔
386
387	spec: GroupSpec	1✔
388
389
390	# endregion

Twingate / kubernetes-operator / 11004195341

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