1import copy
2from typing import TYPE_CHECKING, Any, Dict
3
4from ..config import Config
5from .irresource import IRResource
6
7if TYPE_CHECKING:
8 from .ir import IR # pragma: no cover
9
10
11class IRCORS(IRResource):
12 def __init__(
13 self,
14 ir: "IR",
15 aconf: Config,
16 rkey: str = "ir.cors",
17 kind: str = "IRCORS",
18 name: str = "ir.cors",
19 **kwargs,
20 ) -> None:
21 # print("IRCORS __init__ (%s %s %s)" % (kind, name, kwargs))
22
23 # Convert our incoming kwargs into the things that Envoy actually wants.
24 # Note that we have to treat 'origins' specially here, so that comes after
25 # this renaming loop.
26
27 new_kwargs: Dict[str, Any] = {}
28
29 for from_key, to_key in [
30 ("max_age", "max_age"),
31 ("credentials", "allow_credentials"),
32 ("methods", "allow_methods"),
33 ("headers", "allow_headers"),
34 ("exposed_headers", "expose_headers"),
35 ]:
36 value = kwargs.get(from_key, None)
37
38 if value:
39 new_kwargs[to_key] = self._cors_normalize(value)
40
41 # 'origins' cannot be treated like other keys, because we have to transform it; Envoy wants
42 # it in a different shape than it is in the CRD.
43 origins = kwargs.get("origins", None)
44 if origins is not None:
45 new_kwargs["allow_origin_string_match"] = [{"exact": origin} for origin in origins]
46
47 super().__init__(ir=ir, aconf=aconf, rkey=rkey, kind=kind, name=name, **new_kwargs)
48
49 def setup(self, ir: "IR", aconf: Config) -> bool:
50 # This IRCORS has not been finalized with an ID, so leave with an 'unset' ID so far.
51 self.set_id("unset")
52
53 return True
54
55 def set_id(self, group_id: str):
56 self["filter_enabled"] = {
57 "default_value": {"denominator": "HUNDRED", "numerator": 100},
58 "runtime_key": f"routing.cors_enabled.{group_id}",
59 }
60
61 def dup(self) -> "IRCORS":
62 return copy.copy(self)
63
64 @staticmethod
65 def _cors_normalize(value: Any) -> Any:
66 """
67 List values get turned into a comma-separated string. Other values
68 are returned unaltered.
69 """
70
71 if type(value) == list:
72 return ", ".join([str(x) for x in value])
73 else:
74 return value
75
76 def as_dict(self) -> dict:
77 raw_dict = super().as_dict()
78
79 for key in list(raw_dict):
80 if key in [
81 "_active",
82 "_errored",
83 "_referenced_by",
84 "_rkey",
85 "kind",
86 "location",
87 "name",
88 "namespace",
89 "metadata_labels",
90 ]:
91 raw_dict.pop(key, None)
92
93 return raw_dict
View as plain text