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