1import json
2import requests
3
4class ChallTestServer:
5 """
6 ChallTestServer is a wrapper around pebble-challtestsrv's HTTP management
7 API. If the pebble-challtestsrv process you want to interact with is using
8 a -management argument other than the default ('http://10.77.77.77:8055') you
9 can instantiate the ChallTestServer using the -management address in use. If
10 no custom address is provided the default is assumed.
11 """
12 _baseURL = "http://10.77.77.77:8055"
13
14 _paths = {
15 "set-ipv4": "/set-default-ipv4",
16 "set-ipv6": "/set-default-ipv6",
17 "del-history": "/clear-request-history",
18 "get-http-history": "/http-request-history",
19 "get-dns-history": "/dns-request-history",
20 "get-alpn-history": "/tlsalpn01-request-history",
21 "add-a": "/add-a",
22 "del-a": "/clear-a",
23 "add-aaaa": "/add-aaaa",
24 "del-aaaa": "/clear-aaaa",
25 "add-caa": "/add-caa",
26 "del-caa": "/clear-caa",
27 "add-redirect": "/add-redirect",
28 "del-redirect": "/del-redirect",
29 "add-http": "/add-http01",
30 "del-http": "/del-http01",
31 "add-txt": "/set-txt",
32 "del-txt": "/clear-txt",
33 "add-alpn": "/add-tlsalpn01",
34 "del-alpn": "/del-tlsalpn01",
35 "add-servfail": "/set-servfail",
36 "del-servfail": "/clear-servfail",
37 }
38
39 def __init__(self, url=None):
40 if url is not None:
41 self._baseURL = url
42
43 def _postURL(self, url, body):
44 response = requests.post(
45 url,
46 data=json.dumps(body))
47 return response.text
48
49 def _URL(self, path):
50 urlPath = self._paths.get(path, None)
51 if urlPath is None:
52 raise Exception("No challenge test server URL path known for {0}".format(path))
53 return self._baseURL + urlPath
54
55 def _clear_request_history(self, host, typ):
56 return self._postURL(
57 self._URL("del-history"),
58 { "host": host, "type": typ })
59
60 def set_default_ipv4(self, address):
61 """
62 set_default_ipv4 sets the challenge server's default IPv4 address used
63 to respond to A queries when there are no specific mock A addresses for
64 the hostname being queried. Provide an empty string as the default
65 address to disable answering A queries except for hosts that have mock
66 A addresses added.
67 """
68 return self._postURL(
69 self._URL("set-ipv4"),
70 { "ip": address })
71
72 def set_default_ipv6(self, address):
73 """
74 set_default_ipv6 sets the challenge server's default IPv6 address used
75 to respond to AAAA queries when there are no specific mock AAAA
76 addresses for the hostname being queried. Provide an empty string as the
77 default address to disable answering AAAA queries except for hosts that
78 have mock AAAA addresses added.
79 """
80 return self._postURL(
81 self._URL("set-ipv6"),
82 { "ip": address })
83
84 def add_a_record(self, host, addresses):
85 """
86 add_a_record adds a mock A response to the challenge server's DNS
87 interface for the given host and IPv4 addresses.
88 """
89 return self._postURL(
90 self._URL("add-a"),
91 { "host": host, "addresses": addresses })
92
93 def remove_a_record(self, host):
94 """
95 remove_a_record removes a mock A response from the challenge server's DNS
96 interface for the given host.
97 """
98 return self._postURL(
99 self._URL("del-a"),
100 { "host": host })
101
102 def add_aaaa_record(self, host, addresses):
103 """
104 add_aaaa_record adds a mock AAAA response to the challenge server's DNS
105 interface for the given host and IPv6 addresses.
106 """
107 return self._postURL(
108 self._URL("add-aaaa"),
109 { "host": host, "addresses": addresses })
110
111 def remove_aaaa_record(self, host):
112 """
113 remove_aaaa_record removes mock AAAA response from the challenge server's DNS
114 interface for the given host.
115 """
116 return self._postURL(
117 self._URL("del-aaaa"),
118 { "host": host })
119
120 def add_caa_issue(self, host, value):
121 """
122 add_caa_issue adds a mock CAA response to the challenge server's DNS
123 interface. The mock CAA response will contain one policy with an "issue"
124 tag specifying the provided value.
125 """
126 return self._postURL(
127 self._URL("add-caa"),
128 {
129 "host": host,
130 "policies": [{ "tag": "issue", "value": value}],
131 })
132
133 def remove_caa_issue(self, host):
134 """
135 remove_caa_issue removes a mock CAA response from the challenge server's
136 DNS interface for the given host.
137 """
138 return self._postURL(
139 self._URL("del-caa"),
140 { "host": host })
141
142 def http_request_history(self, host):
143 """
144 http_request_history fetches the challenge server's HTTP request history for the given host.
145 """
146 return json.loads(self._postURL(
147 self._URL("get-http-history"),
148 { "host": host }))
149
150 def clear_http_request_history(self, host):
151 """
152 clear_http_request_history clears the challenge server's HTTP request history for the given host.
153 """
154 return self._clear_request_history(host, "http")
155
156 def add_http_redirect(self, path, targetURL):
157 """
158 add_http_redirect adds a redirect to the challenge server's HTTP
159 interfaces for HTTP requests to the given path directing the client to
160 the targetURL. Redirects are not served for HTTPS requests.
161 """
162 return self._postURL(
163 self._URL("add-redirect"),
164 { "path": path, "targetURL": targetURL })
165
166 def remove_http_redirect(self, path):
167 """
168 remove_http_redirect removes a redirect from the challenge server's HTTP
169 interfaces for the given path.
170 """
171 return self._postURL(
172 self._URL("del-redirect"),
173 { "path": path })
174
175 def add_http01_response(self, token, keyauth):
176 """
177 add_http01_response adds an ACME HTTP-01 challenge response for the
178 provided token under the /.well-known/acme-challenge/ path of the
179 challenge test server's HTTP interfaces. The given keyauth will be
180 returned as the HTTP response body for requests to the challenge token.
181 """
182 return self._postURL(
183 self._URL("add-http"),
184 { "token": token, "content": keyauth })
185
186 def remove_http01_response(self, token):
187 """
188 remove_http01_response removes an ACME HTTP-01 challenge response for
189 the provided token from the challenge test server.
190 """
191 return self._postURL(
192 self._URL("del-http"),
193 { "token": token })
194
195 def add_servfail_response(self, host):
196 """
197 add_servfail_response configures the challenge test server to return
198 SERVFAIL for all queries made for the provided host. This will override
199 any other mocks for the host until removed with remove_servfail_response.
200 """
201 return self._postURL(
202 self._URL("add-servfail"),
203 { "host": host})
204
205 def remove_servfail_response(self, host):
206 """
207 remove_servfail_response undoes the work of add_servfail_response,
208 removing the SERVFAIL configuration for the given host.
209 """
210 return self._postURL(
211 self._URL("del-servfail"),
212 { "host": host})
213
214 def add_dns01_response(self, host, value):
215 """
216 add_dns01_response adds an ACME DNS-01 challenge response for the
217 provided host to the challenge test server's DNS interfaces. The
218 provided value will be served for TXT queries for
219 _acme-challenge.<host>.
220 """
221 if host.endswith(".") is False:
222 host = host + "."
223 return self._postURL(
224 self._URL("add-txt"),
225 { "host": host, "value": value})
226
227 def remove_dns01_response(self, host):
228 """
229 remove_dns01_response removes an ACME DNS-01 challenge response for the
230 provided host from the challenge test server's DNS interfaces.
231 """
232 return self._postURL(
233 self._URL("del-txt"),
234 { "host": host })
235
236 def dns_request_history(self, host):
237 """
238 dns_request_history returns the history of DNS requests made to the
239 challenge test server's DNS interfaces for the given host.
240 """
241 return json.loads(self._postURL(
242 self._URL("get-dns-history"),
243 { "host": host }))
244
245 def clear_dns_request_history(self, host):
246 """
247 clear_dns_request_history clears the history of DNS requests made to the
248 challenge test server's DNS interfaces for the given host.
249 """
250 return self._clear_request_history(host, "dns")
251
252 def add_tlsalpn01_response(self, host, value):
253 """
254 add_tlsalpn01_response adds an ACME TLS-ALPN-01 challenge response
255 certificate to the challenge test server's TLS-ALPN-01 interface for the
256 given host. The provided key authorization value will be embedded in the
257 response certificate served to clients that initiate a TLS-ALPN-01
258 challenge validation with the challenge test server for the provided
259 host.
260 """
261 return self._postURL(
262 self._URL("add-alpn"),
263 { "host": host, "content": value})
264
265 def remove_tlsalpn01_response(self, host):
266 """
267 remove_tlsalpn01_response removes an ACME TLS-ALPN-01 challenge response
268 certificate from the challenge test server's TLS-ALPN-01 interface for
269 the given host.
270 """
271 return self._postURL(
272 self._URL("del-alpn"),
273 { "host": host })
274
275 def tlsalpn01_request_history(self, host):
276 """
277 tls_alpn01_request_history returns the history of TLS-ALPN-01 requests
278 made to the challenge test server's TLS-ALPN-01 interface for the given
279 host.
280 """
281 return json.loads(self._postURL(
282 self._URL("get-alpn-history"),
283 { "host": host }))
284
285 def clear_tlsalpn01_request_history(self, host):
286 """
287 clear_tlsalpn01_request_history clears the history of TLS-ALPN-01
288 requests made to the challenge test server's TLS-ALPN-01 interface for
289 the given host.
290 """
291 return self._clear_request_history(host, "tlsalpn")
View as plain text