1import logging
2
3import pytest
4
5from kat.harness import EDGE_STACK
6from tests.utils import econf_foreach_cluster
7
8logging.basicConfig(
9 level=logging.INFO,
10 format="%(asctime)s test %(levelname)s: %(message)s",
11 datefmt="%Y-%m-%d %H:%M:%S",
12)
13
14logger = logging.getLogger("ambassador")
15
16from ambassador import IR, Config, EnvoyConfig
17from ambassador.fetch import ResourceFetcher
18from ambassador.utils import NullSecretHandler
19from tests.utils import default_listener_manifests
20
21
22def _assert_ext_auth_disabled(route):
23 assert route
24 per_filter_config = route.get("typed_per_filter_config")
25 assert per_filter_config.get("envoy.filters.http.ext_authz")
26 assert per_filter_config.get("envoy.filters.http.ext_authz").get("disabled") == True
27
28
29def _get_ext_auth_config(yaml):
30 for listener in yaml["static_resources"]["listeners"]:
31 for filter_chain in listener["filter_chains"]:
32 for f in filter_chain["filters"]:
33 for http_filter in f["typed_config"]["http_filters"]:
34 if http_filter["name"] == "envoy.filters.http.ext_authz":
35 return http_filter
36 return False
37
38
39def _get_envoy_config(yaml):
40 aconf = Config()
41 fetcher = ResourceFetcher(logger, aconf)
42 fetcher.parse_yaml(default_listener_manifests() + yaml, k8s=True)
43
44 aconf.load_all(fetcher.sorted())
45
46 secret_handler = NullSecretHandler(logger, None, None, "0")
47
48 ir = IR(aconf, file_checker=lambda path: True, secret_handler=secret_handler)
49
50 assert ir
51
52 return EnvoyConfig.generate(ir)
53
54
55@pytest.mark.compilertest
56def test_irauth_grpcservice_version_v2():
57 """Test to ensure that setting protocol_version to will cause an error"""
58
59 yaml = """
60---
61apiVersion: getambassador.io/v3alpha1
62kind: AuthService
63metadata:
64 name: mycoolauthservice
65 namespace: default
66spec:
67 auth_service: someservice
68 protocol_version: "v2"
69 proto: grpc
70"""
71
72 econf = _get_envoy_config(yaml)
73
74 conf = econf.as_dict()
75 ext_auth_config = _get_ext_auth_config(conf)
76
77 assert ext_auth_config == False
78
79 errors = econf.ir.aconf.errors["mycoolauthservice.default.1"]
80 assert (
81 errors[0]["error"]
82 == 'AuthService: protocol_version v2 is unsupported, protocol_version must be "v3"'
83 )
84
85
86def test_irauth_grpcservice_version_v3():
87 yaml = """
88---
89apiVersion: getambassador.io/v3alpha1
90kind: AuthService
91metadata:
92 name: mycoolauthservice
93 namespace: default
94spec:
95 auth_service: someservice
96 protocol_version: "v3"
97 proto: grpc
98"""
99
100 econf = _get_envoy_config(yaml)
101
102 conf = econf.as_dict()
103 ext_auth_config = _get_ext_auth_config(conf)
104
105 assert ext_auth_config
106 assert (
107 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
108 == "cluster_extauth_someservice_default"
109 )
110 assert ext_auth_config["typed_config"]["transport_api_version"] == "V3"
111
112 assert "mycoolauthservice.default.1" not in econf.ir.aconf.errors
113
114
115def test_cluster_fields():
116 yaml = """
117---
118apiVersion: getambassador.io/v3alpha1
119kind: AuthService
120metadata:
121 name: mycoolauthservice
122 namespace: default
123spec:
124 auth_service: someservice
125 protocol_version: "v3"
126 proto: grpc
127 stats_name: authservice
128"""
129
130 econf = _get_envoy_config(yaml)
131
132 conf = econf.as_dict()
133 ext_auth_config = _get_ext_auth_config(conf)
134
135 cluster_name = "cluster_extauth_someservice_default"
136
137 assert ext_auth_config
138 assert (
139 ext_auth_config["typed_config"]["grpc_service"]["envoy_grpc"]["cluster_name"]
140 == cluster_name
141 )
142
143 def check_fields(cluster):
144 assert cluster["alt_stat_name"] == "authservice"
145
146 econf_foreach_cluster(econf.as_dict(), check_fields, name=cluster_name)
147
148
149@pytest.mark.compilertest
150def test_irauth_grpcservice_version_default():
151 if EDGE_STACK:
152 pytest.xfail("XFailing for now, custom AuthServices not supported in Edge Stack")
153
154 yaml = """
155---
156apiVersion: getambassador.io/v3alpha1
157kind: AuthService
158metadata:
159 name: mycoolauthservice
160 namespace: default
161spec:
162 auth_service: someservice
163 proto: grpc
164"""
165
166 econf = _get_envoy_config(yaml)
167
168 conf = econf.as_dict()
169 ext_auth_config = _get_ext_auth_config(conf)
170
171 assert ext_auth_config == False
172
173 errors = econf.ir.aconf.errors["mycoolauthservice.default.1"]
174 assert (
175 errors[0]["error"]
176 == 'AuthService: protocol_version v2 is unsupported, protocol_version must be "v3"'
177 )
178
179
180@pytest.mark.compilertest
181def test_basic_http_redirect_with_no_authservice():
182 """Test that http --> https redirect route exists when no AuthService is provided
183 and verify that the typed_per_filter_config is NOT included
184 """
185
186 yaml = """
187apiVersion: getambassador.io/v3alpha1
188kind: Mapping
189metadata:
190 name: ambassador
191 namespace: default
192spec:
193 hostname: "*"
194 prefix: /httpbin/
195 service: httpbin
196 """
197 econf = _get_envoy_config(yaml)
198
199 for rv in econf.route_variants:
200 if rv.route.get("match").get("prefix") == "/httpbin/":
201 xfp_http_redirect = rv.variants.get("xfp-http-redirect")
202 assert xfp_http_redirect
203 assert "redirect" in xfp_http_redirect
204 assert "typed_per_filter_config" not in xfp_http_redirect
205
206
207@pytest.mark.compilertest
208def test_redirects_disables_ext_authz():
209 """Test that the ext_authz is disabled on envoy redirect routes
210 for https_redirects and host_redirects. This is to ensure that the
211 redirect occurs before making any calls to the ext_authz service
212 """
213
214 if EDGE_STACK:
215 pytest.xfail("XFailing for now, custom AuthServices not supported in Edge Stack")
216
217 yaml = """
218---
219apiVersion: getambassador.io/v3alpha1
220kind: AuthService
221metadata:
222 name: mycoolauthservice
223 namespace: default
224spec:
225 auth_service: someservice
226 proto: grpc
227 protocol_version: v3
228---
229apiVersion: getambassador.io/v3alpha1
230kind: Mapping
231metadata:
232 name: ambassador
233 namespace: default
234spec:
235 hostname: "*"
236 prefix: /httpbin/
237 service: httpbin
238 host_redirect: true
239 """
240 econf = _get_envoy_config(yaml)
241
242 # check https_redirect variant route
243 for rv in econf.route_variants:
244 if rv.route.get("match").get("prefix") == "/httpbin/":
245 xfp_http_redirect = rv.variants.get("xfp-http-redirect")
246 assert xfp_http_redirect
247 assert "redirect" in xfp_http_redirect
248 _assert_ext_auth_disabled(xfp_http_redirect)
249
250 # check host_redirect route
251 for route in econf.routes:
252 if route.get("match").get("prefix") == "/httpbin/":
253 redirect = route.get("redirect")
254 assert redirect
255 assert redirect.get("host_redirect") == "httpbin"
256 _assert_ext_auth_disabled(route)
View as plain text