Coverage for webapp/handlers.py: 87%
148 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-08 12:33 +0000
1import socket
2from urllib.parse import unquote, urlparse, urlunparse
4import base64
5import hashlib
6import re
7import sentry_sdk
9import flask
10from flask import render_template, request
11import webapp.template_utils as template_utils
12from canonicalwebteam import image_template
13from webapp import authentication
14import webapp.helpers as helpers
15from webapp.config import (
16 BSI_URL,
17 LOGIN_URL,
18 SENTRY_DSN,
19 COMMIT_ID,
20 ENVIRONMENT,
21 WEBAPP_CONFIG,
22 DNS_VERIFICATION_SALT,
23 IS_DEVELOPMENT,
24 VITE_PORT,
25 DEFAULT_ICON_URL,
26 STATUS_BANNER,
27)
29from canonicalwebteam.exceptions import (
30 StoreApiError,
31 StoreApiConnectionError,
32 StoreApiResourceNotFound,
33 StoreApiResponseDecodeError,
34 StoreApiResponseError,
35 StoreApiResponseErrorList,
36 StoreApiTimeoutError,
37 PublisherAgreementNotSigned,
38 PublisherMacaroonRefreshRequired,
39 PublisherMissingUsername,
40)
42from webapp.api.exceptions import (
43 ApiError,
44 ApiConnectionError,
45 ApiResponseErrorList,
46 ApiTimeoutError,
47 ApiResponseDecodeError,
48 ApiResponseError,
49)
51from datetime import datetime
53CSP = {
54 "default-src": ["'self'"],
55 "img-src": [
56 "data: blob:",
57 # This is needed to allow images from
58 # https://www.google.*/ads/ga-audiences to load.
59 "*",
60 ],
61 "script-src-elem": [
62 "'self'",
63 "assets.ubuntu.com",
64 "www.googletagmanager.com",
65 "www.youtube.com",
66 "asciinema.org",
67 "player.vimeo.com",
68 "plausible.io",
69 "script.crazyegg.com",
70 "w.usabilla.com",
71 "connect.facebook.net",
72 "snap.licdn.com",
73 "challenges.cloudflare.com",
74 # This is necessary for Google Tag Manager to function properly.
75 "'unsafe-inline'",
76 ],
77 "font-src": [
78 "'self'",
79 "assets.ubuntu.com",
80 ],
81 "script-src": [],
82 "connect-src": [
83 "'self'",
84 "ubuntu.com",
85 "analytics.google.com",
86 "*.analytics.google.com",
87 "stats.g.doubleclick.net",
88 "www.googletagmanager.com",
89 "sentry.is.canonical.com",
90 "www.google-analytics.com",
91 "plausible.io",
92 "*.crazyegg.com",
93 "www.facebook.com",
94 "px.ads.linkedin.com",
95 "*.snapcraft.io",
96 "*.snapcraftcontent.com",
97 "marketplace-analytics.staging.canonical.com",
98 "marketplace-analytics.canonical.com",
99 "challenges.cloudflare.com",
100 "www.google.com",
101 ],
102 "frame-src": [
103 "'self'",
104 "td.doubleclick.net",
105 "www.youtube.com",
106 "youtube.com",
107 "asciinema.org",
108 "player.vimeo.com",
109 "snapcraft.io",
110 "www.facebook.com",
111 "challenges.cloudflare.com",
112 "snap:",
113 ],
114 "style-src": [
115 "'self'",
116 "'unsafe-inline'",
117 ],
118 "media-src": [
119 "'self'",
120 "res.cloudinary.com",
121 ],
122}
124CSP_SCRIPT_SRC = [
125 "'self'",
126 "blob:",
127 "'unsafe-eval'",
128 "'unsafe-hashes'",
129]
131# Vite integration
132if IS_DEVELOPMENT:
133 CSP["script-src-elem"].append(f"localhost:{VITE_PORT}")
134 CSP["connect-src"].append(f"localhost:{VITE_PORT}")
135 CSP["connect-src"].append(f"ws://localhost:{VITE_PORT}")
136 CSP["style-src"].append(f"localhost:{VITE_PORT}")
137 CSP_SCRIPT_SRC.append(f"localhost:{VITE_PORT}")
140def refresh_redirect():
141 if "macaroon_exchanged" in flask.session:
142 authentication.reset_auth_session(flask.session)
143 return flask.redirect(
144 flask.url_for(
145 "login.login_handler",
146 next=flask.request.full_path.rstrip("?"),
147 )
148 )
150 try:
151 macaroon_discharge = authentication.get_refreshed_discharge(
152 flask.session["macaroon_discharge"]
153 )
154 except ApiResponseError as api_response_error:
155 if api_response_error.status_code == 401:
156 return flask.redirect(flask.url_for("login.logout"))
157 else:
158 return flask.abort(502, str(api_response_error))
159 except ApiError as api_error:
160 return flask.abort(502, str(api_error))
162 flask.session["macaroon_discharge"] = macaroon_discharge
163 return flask.redirect(
164 flask.url_for(
165 flask.request.endpoint,
166 **flask.request.view_args,
167 **flask.request.args,
168 )
169 )
172def snapcraft_utility_processor():
173 if authentication.is_authenticated(flask.session):
174 user_name = flask.session["publisher"]["fullname"]
175 user_is_canonical = flask.session["publisher"].get(
176 "is_canonical", False
177 )
178 stores = flask.session["publisher"].get("stores")
179 else:
180 user_name = None
181 user_is_canonical = False
182 stores = []
184 page_slug = template_utils.generate_slug(flask.request.path)
186 return {
187 # Variables
188 "LOGIN_URL": LOGIN_URL,
189 "SENTRY_DSN": SENTRY_DSN,
190 "COMMIT_ID": COMMIT_ID,
191 "ENVIRONMENT": ENVIRONMENT,
192 "host_url": flask.request.host_url,
193 "path": flask.request.path,
194 "page_slug": page_slug,
195 "user_name": user_name,
196 "VERIFIED_PUBLISHER": "verified",
197 "STAR_DEVELOPER": "starred",
198 "webapp_config": WEBAPP_CONFIG,
199 "BSI_URL": BSI_URL,
200 "now": datetime.now(),
201 "user_is_canonical": user_is_canonical,
202 # Functions
203 "contains": template_utils.contains,
204 "join": template_utils.join,
205 "static_url": template_utils.static_url,
206 "IS_DEVELOPMENT": IS_DEVELOPMENT,
207 "format_number": template_utils.format_number,
208 "format_display_name": template_utils.format_display_name,
209 "display_name": template_utils.display_name,
210 "install_snippet": template_utils.install_snippet,
211 "format_date": template_utils.format_date,
212 "format_member_role": template_utils.format_member_role,
213 "image": image_template,
214 "stores": stores,
215 "format_link": template_utils.format_link,
216 "DNS_VERIFICATION_SALT": DNS_VERIFICATION_SALT,
217 "DEFAULT_ICON_URL": DEFAULT_ICON_URL,
218 "STATUS_BANNER": STATUS_BANNER,
219 }
222def set_handlers(app):
223 @app.context_processor
224 def utility_processor():
225 """
226 This defines the set of properties and functions that will be added
227 to the default context for processing templates. All these items
228 can be used in all templates
229 """
231 return snapcraft_utility_processor()
233 # Error handlers
234 # ===
235 @app.errorhandler(500)
236 @app.errorhandler(501)
237 @app.errorhandler(502)
238 @app.errorhandler(504)
239 @app.errorhandler(505)
240 def internal_error(error):
241 error_name = getattr(error, "name", type(error).__name__)
242 return_code = getattr(error, "code", 500)
244 if not app.testing:
245 sentry_sdk.capture_exception()
247 return (
248 flask.render_template("50X.html", error_name=error_name),
249 return_code,
250 )
252 @app.errorhandler(503)
253 def service_unavailable(error):
254 return render_template("503.html"), 503
256 @app.errorhandler(404)
257 @app.errorhandler(StoreApiResourceNotFound)
258 def handle_resource_not_found(error):
259 return render_template("404.html", message=str(error)), 404
261 @app.errorhandler(ApiTimeoutError)
262 @app.errorhandler(StoreApiTimeoutError)
263 def handle_connection_timeout(error):
264 status_code = 504
265 return (
266 render_template(
267 "50X.html", error_message=str(error), status_code=status_code
268 ),
269 status_code,
270 )
272 @app.errorhandler(ApiResponseDecodeError)
273 @app.errorhandler(ApiResponseError)
274 @app.errorhandler(ApiConnectionError)
275 @app.errorhandler(StoreApiResponseDecodeError)
276 @app.errorhandler(StoreApiResponseError)
277 @app.errorhandler(StoreApiConnectionError)
278 @app.errorhandler(ApiError)
279 @app.errorhandler(StoreApiError)
280 def store_api_error(error):
281 status_code = 502
282 return (
283 render_template(
284 "50X.html", error_message=str(error), status_code=status_code
285 ),
286 status_code,
287 )
289 @app.errorhandler(ApiResponseErrorList)
290 @app.errorhandler(StoreApiResponseErrorList)
291 def handle_api_error_list(error):
292 if error.status_code == 404:
293 if "snap_name" in request.path:
294 return flask.abort(404, "Snap not found!")
295 else:
296 return (
297 render_template("404.html", message="Entity not found"),
298 404,
299 )
300 if len(error.errors) == 1 and error.errors[0]["code"] in [
301 "macaroon-permission-required",
302 "macaroon-authorization-required",
303 ]:
304 authentication.reset_auth_session(flask.session)
305 return flask.redirect(
306 flask.url_for("login.login_handler", next=flask.request.path)
307 )
309 status_code = 502
310 codes = [
311 f"{error['code']}: {error.get('message', 'No message')}"
312 for error in error.errors
313 ]
315 error_msg = ", ".join(codes)
316 return (
317 render_template(
318 "50X.html", error_message=error_msg, status_code=status_code
319 ),
320 status_code,
321 )
323 # Publisher error
324 @app.errorhandler(PublisherMissingUsername)
325 def handle_publisher_missing_name(error):
326 return flask.redirect(flask.url_for("account.get_account_name"))
328 @app.errorhandler(PublisherAgreementNotSigned)
329 def handle_publisher_agreement_not_signed(error):
330 return flask.redirect(flask.url_for("account.get_agreement"))
332 @app.errorhandler(PublisherMacaroonRefreshRequired)
333 def handle_publisher_macaroon_refresh_required(error):
334 return refresh_redirect()
336 # Global tasks for all requests
337 # ===
338 @app.before_request
339 def clear_trailing():
340 """
341 Remove trailing slashes from all routes
342 We like our URLs without slashes
343 """
345 parsed_url = urlparse(unquote(flask.request.url))
346 path = parsed_url.path
348 if path != "/" and path.endswith("/"):
349 new_uri = urlunparse(parsed_url._replace(path=path[:-1]))
351 return flask.redirect(new_uri)
353 # Calculate the SHA256 hash of the script content and encode it in base64.
354 def calculate_sha256_base64(script_content):
355 sha256_hash = hashlib.sha256(script_content.encode()).digest()
356 return "sha256-" + base64.b64encode(sha256_hash).decode()
358 def get_csp_directive(content, regex):
359 directive_items = set()
360 pattern = re.compile(regex)
361 matched_contents = pattern.findall(content)
362 for matched_content in matched_contents:
363 hash_value = f"'{calculate_sha256_base64(matched_content)}'"
364 directive_items.add(hash_value)
365 return list(directive_items)
367 # Find all script elements in the response and add their hashes to the CSP.
368 def add_script_hashes_to_csp(response):
369 response.freeze()
370 decoded_content = b"".join(response.response).decode(
371 "utf-8", errors="replace"
372 )
374 CSP["script-src"] = CSP_SCRIPT_SRC + get_csp_directive(
375 decoded_content, r'onclick\s*=\s*"(.*?)"'
376 )
377 return CSP
379 @app.after_request
380 def add_headers(response):
381 """
382 Generic rules for headers to add to all requests
384 - X-Hostname: Mention the name of the host/pod running the application
385 - Cache-Control: Add cache-control headers for public and private pages
386 - Content-Security-Policy: Restrict resources (e.g., JavaScript, CSS,
387 Images) and URLs
388 - Referrer-Policy: Limit referrer data for security while preserving
389 full referrer for same-origin requests
390 - Cross-Origin-Embedder-Policy: allows embedding cross-origin
391 resources
392 - Cross-Origin-Opener-Policy: enable the page to open pop-ups while
393 maintaining same-origin policy
394 - Cross-Origin-Resource-Policy: allowing cross-origin requests to
395 access the resource
396 - X-Permitted-Cross-Domain-Policies: disallows cross-domain access to
397 resources
398 """
400 response.headers["X-Hostname"] = socket.gethostname()
402 if response.status_code == 200:
403 if flask.session:
404 response.headers["Cache-Control"] = "private"
405 else:
406 # Only add caching headers to successful responses
407 if not response.headers.get("Cache-Control"):
408 response.headers["Cache-Control"] = ", ".join(
409 {
410 "public",
411 "max-age=61",
412 "stale-while-revalidate=300",
413 "stale-if-error=86400",
414 }
415 )
416 csp = add_script_hashes_to_csp(response)
417 response.headers["Content-Security-Policy"] = helpers.get_csp_as_str(
418 csp
419 )
420 response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
421 response.headers["Cross-Origin-Embedder-Policy"] = "unsafe-none"
422 response.headers["Cross-Origin-Opener-Policy"] = (
423 "same-origin-allow-popups"
424 )
425 response.headers["Cross-Origin-Resource-Policy"] = "cross-origin"
426 response.headers["X-Permitted-Cross-Domain-Policies"] = "none"
427 return response