Coverage for webapp/store/snap_details_views.py: 84%

224 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-25 22:09 +0000

1import flask 

2from flask import Response 

3import requests 

4 

5import logging 

6import humanize 

7import os 

8 

9import webapp.helpers as helpers 

10import webapp.metrics.helper as metrics_helper 

11import webapp.metrics.metrics as metrics 

12import webapp.store.logic as logic 

13from webapp import authentication 

14from webapp.markdown import parse_markdown_description 

15from cache.cache_utility import redis_cache 

16 

17from canonicalwebteam.flask_base.decorators import ( 

18 exclude_xframe_options_header, 

19) 

20from canonicalwebteam.exceptions import StoreApiError, StoreApiResourceNotFound 

21from canonicalwebteam.store_api.devicegw import DeviceGW 

22from pybadges import badge 

23 

24device_gateway = DeviceGW("snap", helpers.api_session) 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29FIELDS = [ 

30 "title", 

31 "summary", 

32 "description", 

33 "license", 

34 "contact", 

35 "website", 

36 "publisher", 

37 "media", 

38 "download", 

39 "version", 

40 "created-at", 

41 "confinement", 

42 "categories", 

43 "trending", 

44 "unlisted", 

45 "links", 

46 "revision", 

47 "sboms", 

48] 

49 

50FIELDS_EXTRA_DETAILS = [ 

51 "aliases", 

52] 

53 

54 

55def snap_details_views(store): 

56 snap_regex = "[a-z0-9-]*[a-z][a-z0-9-]*" 

57 snap_regex_upercase = "[A-Za-z0-9-]*[A-Za-z][A-Za-z0-9-]*" 

58 

59 def _get_context_snap_details(snap_name, supported_architectures=None): 

60 details = device_gateway.get_item_details( 

61 snap_name, fields=FIELDS, api_version=2 

62 ) 

63 # 404 for any snap under quarantine 

64 if details["snap"]["publisher"]["username"] == "snap-quarantine": 

65 flask.abort(404, "No snap named {}".format(snap_name)) 

66 

67 # When removing all the channel maps of an existing snap the API, 

68 # responds that the snaps still exists with data. 

69 # Return a 404 if not channel maps, to avoid having a error. 

70 # For example: mir-kiosk-browser 

71 if not details.get("channel-map"): 

72 flask.abort(404, "No snap named {}".format(snap_name)) 

73 

74 formatted_description = parse_markdown_description( 

75 details.get("snap", {}).get("description", "") 

76 ) 

77 

78 channel_maps_list = logic.convert_channel_maps( 

79 details.get("channel-map") 

80 ) 

81 

82 latest_channel = logic.get_last_updated_version( 

83 details.get("channel-map") 

84 ) 

85 

86 revisions = logic.get_revisions(details.get("channel-map")) 

87 

88 default_track = ( 

89 details.get("default-track") 

90 if details.get("default-track") 

91 else "latest" 

92 ) 

93 

94 lowest_risk_available = logic.get_lowest_available_risk( 

95 channel_maps_list, default_track 

96 ) 

97 

98 extracted_info = logic.extract_info_channel_map( 

99 channel_maps_list, default_track, lowest_risk_available 

100 ) 

101 

102 last_updated = latest_channel["channel"]["released-at"] 

103 updates = logic.get_latest_versions( 

104 details.get("channel-map"), 

105 default_track, 

106 lowest_risk_available, 

107 supported_architectures, 

108 ) 

109 

110 # Determine the most recent update date from updates tuple 

111 # updates[0] is the stable channel, updates[1] is the most 

112 # recent non-stable 

113 most_recent_update = None 

114 if updates[0] and updates[1]: 

115 # Compare both and use the most recent 

116 date_0 = updates[0].get("released-at") 

117 date_1 = updates[1].get("released-at") 

118 if date_0 and date_1: 

119 most_recent_update = max(date_0, date_1) 

120 else: 

121 most_recent_update = date_0 or date_1 

122 elif updates[0]: 

123 most_recent_update = updates[0].get("released-at") 

124 elif updates[1]: 

125 most_recent_update = updates[1].get("released-at") 

126 

127 binary_filesize = latest_channel["download"]["size"] 

128 

129 # filter out banner and banner-icon images from screenshots 

130 screenshots = logic.filter_screenshots( 

131 details.get("snap", {}).get("media", []) 

132 ) 

133 

134 icon_url = helpers.get_icon(details.get("snap", {}).get("media", [])) 

135 

136 publisher_info = helpers.get_yaml( 

137 "{}{}.yaml".format( 

138 flask.current_app.config["CONTENT_DIRECTORY"][ 

139 "PUBLISHER_PAGES" 

140 ], 

141 details["snap"]["publisher"]["username"], 

142 ), 

143 typ="safe", 

144 ) 

145 

146 publisher_snaps = [] 

147 publisher_featured_snaps = None 

148 

149 if publisher_info: 

150 publisher_results = logic.get_publisher_snaps( 

151 device_gateway, details["snap"]["publisher"]["username"] 

152 ) 

153 

154 snaps_by_name = {} 

155 for snap in publisher_results: 

156 item = snap["snap"] 

157 snaps_by_name[snap["name"]] = { 

158 "package_name": snap["name"], 

159 "title": item.get("title"), 

160 "summary": item.get("summary"), 

161 "icon_url": helpers.get_icon(item.get("media", [])), 

162 } 

163 

164 publisher_featured_snaps = logic.hydrate_featured_snaps( 

165 publisher_info.get("featured_snaps"), snaps_by_name 

166 ) 

167 

168 # The "More from publisher" list excludes featured snaps and 

169 # the snap currently being viewed. 

170 excluded_names = {snap_name} 

171 for snap in publisher_featured_snaps: 

172 excluded_names.add(snap["package_name"]) 

173 

174 available_snaps = [ 

175 snap 

176 for name, snap in snaps_by_name.items() 

177 if name not in excluded_names 

178 ] 

179 

180 publisher_snaps = logic.get_n_random_snaps(available_snaps, 4) 

181 

182 video = logic.get_video(details.get("snap", {}).get("media", [])) 

183 

184 is_users_snap = False 

185 if authentication.is_authenticated(flask.session): 

186 if ( 

187 flask.session.get("publisher").get("nickname") 

188 == details["snap"]["publisher"]["username"] 

189 ): 

190 is_users_snap = True 

191 

192 # build list of categories of a snap 

193 categories = logic.get_snap_categories( 

194 details.get("snap", {}).get("categories", []) 

195 ) 

196 

197 developer = logic.get_snap_developer(details["name"]) 

198 

199 is_last_updated_old = logic.is_snap_old(last_updated) 

200 

201 context = { 

202 "snap_id": details.get("snap-id"), 

203 # Data direct from details API 

204 "snap_title": details["snap"]["title"], 

205 "package_name": details["name"], 

206 "categories": categories, 

207 "icon_url": icon_url, 

208 "version": extracted_info["version"], 

209 "license": details["snap"]["license"], 

210 "publisher": details["snap"]["publisher"]["display-name"], 

211 "username": details["snap"]["publisher"]["username"], 

212 "screenshots": screenshots, 

213 "video": video, 

214 "publisher_snaps": publisher_snaps, 

215 "publisher_featured_snaps": publisher_featured_snaps, 

216 "has_publisher_page": publisher_info is not None, 

217 "contact": details["snap"].get("contact"), 

218 "website": details["snap"].get("website"), 

219 "summary": details["snap"]["summary"], 

220 "description": formatted_description, 

221 "channel_map": channel_maps_list, 

222 "has_stable": logic.has_stable(channel_maps_list), 

223 "developer_validation": details["snap"]["publisher"]["validation"], 

224 "default_track": default_track, 

225 "lowest_risk_available": lowest_risk_available, 

226 "confinement": extracted_info["confinement"], 

227 "trending": details.get("snap", {}).get("trending", False), 

228 # Transformed API data 

229 "filesize": humanize.naturalsize(binary_filesize), 

230 "last_updated": logic.convert_date(last_updated), 

231 "last_updated_raw": last_updated, 

232 "is_snap_old": logic.is_snap_old(most_recent_update), 

233 "is_last_updated_old": is_last_updated_old, 

234 "is_users_snap": is_users_snap, 

235 "unlisted": details.get("snap", {}).get("unlisted", False), 

236 "developer": developer, 

237 # TODO: This is horrible and hacky 

238 "appliances": { 

239 "adguard-home": "adguard", 

240 "mosquitto": "mosquitto", 

241 "nextcloud": "nextcloud", 

242 "plexmediaserver": "plex", 

243 "openhab": "openhab", 

244 }, 

245 "links": details["snap"].get("links"), 

246 "updates": updates, 

247 "revisions": revisions, 

248 "turnstile_site_key": ( 

249 flask.current_app.config.get("TURNSTILE_SITE_KEY", "") 

250 if flask.current_app.config.get("TURNSTILE_SECRET_KEY") 

251 else "" 

252 ), 

253 } 

254 return context 

255 

256 def verify_turnstile(turnstile_response): 

257 turnstile_secret = flask.current_app.config.get( 

258 "TURNSTILE_SECRET_KEY", "" 

259 ) 

260 if not turnstile_secret: 

261 return True 

262 

263 if not turnstile_response: 

264 logger.warning("Turnstile token missing from report form") 

265 return False 

266 

267 payload = { 

268 "secret": turnstile_secret, 

269 "response": turnstile_response, 

270 } 

271 

272 try: 

273 response = requests.post( 

274 flask.current_app.config["TURNSTILE_VERIFY_URL"], 

275 data=payload, 

276 timeout=10, 

277 ) 

278 if not response.ok: 

279 logger.warning( 

280 "Turnstile verification returned %s", 

281 response.status_code, 

282 ) 

283 return False 

284 verification = response.json() 

285 except (requests.RequestException, ValueError): 

286 logger.exception("Turnstile verification failed") 

287 return False 

288 

289 if not verification.get("success"): 

290 logger.warning( 

291 "Turnstile verification denied report: %s", 

292 verification.get("error-codes", []), 

293 ) 

294 return False 

295 

296 return True 

297 

298 @store.route('/<regex("' + snap_regex + '"):snap_name>') 

299 def snap_details(snap_name): 

300 """ 

301 A view to display the snap details page for specific snaps. 

302 

303 This queries the snapcraft API (api.snapcraft.io) and passes 

304 some of the data through to the snap-details.html template, 

305 with appropriate sanitation. 

306 """ 

307 

308 error_info = {} 

309 status_code = 200 

310 

311 context = _get_context_snap_details(snap_name) 

312 try: 

313 # the empty string channel makes the store API not filter by 

314 # the default channel 'latest/stable', which gives errors for 

315 # snaps that don't use that channel 

316 extra_details = device_gateway.get_snap_details( 

317 snap_name, channel="", fields=FIELDS_EXTRA_DETAILS 

318 ) 

319 except StoreApiResourceNotFound: 

320 # not all snaps are served by the details endpoint, for 

321 # example gated snaps, and the page renders without aliases 

322 extra_details = None 

323 except Exception: 

324 logger.exception("Details endpoint returned an error") 

325 extra_details = None 

326 

327 if extra_details and extra_details["aliases"]: 

328 context["aliases"] = [ 

329 [ 

330 f"{extra_details['package_name']}.{alias_obj['target']}", 

331 alias_obj["name"], 

332 ] 

333 for alias_obj in extra_details["aliases"] 

334 ] 

335 

336 country_metric_name = "weekly_installed_base_by_country_percent" 

337 os_metric_name = "weekly_installed_base_by_operating_system_normalized" 

338 

339 end = metrics_helper.get_last_metrics_processed_date() 

340 

341 metrics_query_json = [ 

342 metrics_helper.get_filter( 

343 metric_name=country_metric_name, 

344 snap_id=context["snap_id"], 

345 start=end, 

346 end=end, 

347 ), 

348 metrics_helper.get_filter( 

349 metric_name=os_metric_name, 

350 snap_id=context["snap_id"], 

351 start=end, 

352 end=end, 

353 ), 

354 ] 

355 

356 metrics_response = device_gateway.get_public_metrics( 

357 metrics_query_json 

358 ) 

359 

360 os_metrics = None 

361 country_devices = None 

362 if metrics_response: 

363 oses = metrics_helper.find_metric(metrics_response, os_metric_name) 

364 os_metrics = metrics.OsMetric( 

365 name=oses["metric_name"], 

366 series=oses["series"], 

367 buckets=oses["buckets"], 

368 status=oses["status"], 

369 ) 

370 

371 territories = metrics_helper.find_metric( 

372 metrics_response, country_metric_name 

373 ) 

374 country_devices = metrics.CountryDevices( 

375 name=territories["metric_name"], 

376 series=territories["series"], 

377 buckets=territories["buckets"], 

378 status=territories["status"], 

379 private=False, 

380 ) 

381 

382 context.update( 

383 { 

384 "countries": ( 

385 country_devices.country_data if country_devices else None 

386 ), 

387 "normalized_os": os_metrics.os if os_metrics else None, 

388 # Context info 

389 "is_linux": ( 

390 "Linux" in flask.request.headers.get("User-Agent", "") 

391 and "Android" 

392 not in flask.request.headers.get("User-Agent", "") 

393 ), 

394 "error_info": error_info, 

395 } 

396 ) 

397 

398 context["default_arch"] = logic.get_default_architecture( 

399 context["channel_map"].keys() 

400 ) 

401 

402 return ( 

403 flask.render_template("store/snap-details.html", **context), 

404 status_code, 

405 ) 

406 

407 @store.route('/<regex("' + snap_regex + '"):snap_name>/embedded') 

408 @exclude_xframe_options_header 

409 def snap_details_embedded(snap_name): 

410 """ 

411 A view to display the snap embedded card for specific snaps. 

412 

413 This queries the snapcraft API (api.snapcraft.io) and passes 

414 some of the data through to the template, 

415 with appropriate sanitation. 

416 """ 

417 status_code = 200 

418 

419 context = _get_context_snap_details(snap_name) 

420 

421 button_variants = ["black", "white"] 

422 button = flask.request.args.get("button") 

423 if button and button not in button_variants: 

424 button = "black" 

425 

426 architectures = list(context["channel_map"].keys()) 

427 

428 context.update( 

429 { 

430 "default_architecture": ( 

431 "amd64" if "amd64" in architectures else architectures[0] 

432 ), 

433 "button": button, 

434 "show_channels": flask.request.args.get("channels"), 

435 "show_summary": flask.request.args.get("summary"), 

436 "show_screenshot": flask.request.args.get("screenshot"), 

437 } 

438 ) 

439 

440 return ( 

441 flask.render_template("store/snap-embedded-card.html", **context), 

442 status_code, 

443 ) 

444 

445 @store.route('/<regex("' + snap_regex_upercase + '"):snap_name>') 

446 def snap_details_case_sensitive(snap_name): 

447 return flask.redirect( 

448 flask.url_for(".snap_details", snap_name=snap_name.lower()) 

449 ) 

450 

451 def get_badge_svg(snap_name, left_text, right_text, color="#0e8420"): 

452 show_name = flask.request.args.get("name", default=1, type=int) 

453 snap_link = flask.request.url_root + snap_name 

454 

455 svg = badge( 

456 left_text=left_text if show_name else "", 

457 right_text=right_text, 

458 right_color=color, 

459 left_link=snap_link, 

460 right_link=snap_link, 

461 logo=( 

462 "data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' " 

463 "viewBox='0 0 32 32'%3E%3Cdefs%3E%3Cstyle%3E.cls-1%7Bfill:%23f" 

464 "ff%7D%3C/style%3E%3C/defs%3E%3Cpath class='cls-1' d='M18.03 1" 

465 "8.03l5.95-5.95-5.95-2.65v8.6zM6.66 29.4l10.51-10.51-3.21-3.18" 

466 "-7.3 13.69zM2.5 3.6l15.02 14.94V9.03L2.5 3.6zM27.03 9.03h-8.6" 

467 "5l11.12 4.95-2.47-4.95z'/%3E%3C/svg%3E" 

468 ), 

469 ) 

470 return svg 

471 

472 @store.route('/<regex("' + snap_regex + '"):snap_name>/badge.svg') 

473 def snap_details_badge(snap_name): 

474 context = _get_context_snap_details(snap_name) 

475 

476 # channel with safest risk available in default track 

477 snap_channel = "".join( 

478 [context["default_track"], "/", context["lowest_risk_available"]] 

479 ) 

480 

481 svg = get_badge_svg( 

482 snap_name=snap_name, 

483 left_text=context["snap_title"], 

484 right_text=snap_channel + " " + context["version"], 

485 ) 

486 

487 return svg, 200, {"Content-Type": "image/svg+xml"} 

488 

489 @store.route("/<lang>/<theme>/install.svg") 

490 def snap_install_badge(lang, theme): 

491 base_path = "static/images/badges/" 

492 allowed_langs = helpers.list_folders(base_path) 

493 

494 if lang not in allowed_langs: 

495 return Response("Invalid language", status=400) 

496 

497 file_name = ( 

498 "snap-store-white.svg" 

499 if theme == "light" 

500 else "snap-store-black.svg" 

501 ) 

502 

503 svg_path = os.path.normpath(os.path.join(base_path, lang, file_name)) 

504 

505 # Ensure the path is within the base path 

506 if not svg_path.startswith(base_path) or not os.path.exists(svg_path): 

507 return Response( 

508 '<svg height="20" width="1" ' 

509 'xmlns="http://www.w3.org/2000/svg" ' 

510 'xmlns:xlink="http://www.w3.org/1999/xlink"></svg>', 

511 mimetype="image/svg+xml", 

512 status=404, 

513 ) 

514 else: 

515 with open(svg_path, "r") as svg_file: 

516 svg_content = svg_file.read() 

517 return Response(svg_content, mimetype="image/svg+xml") 

518 

519 @store.route('/<regex("' + snap_regex + '"):snap_name>/trending.svg') 

520 def snap_details_badge_trending(snap_name): 

521 is_preview = flask.request.args.get("preview", default=0, type=int) 

522 context = _get_context_snap_details(snap_name) 

523 

524 # default to empty SVG 

525 svg = ( 

526 '<svg height="20" width="1" xmlns="http://www.w3.org/2000/svg" ' 

527 'xmlns:xlink="http://www.w3.org/1999/xlink"></svg>' 

528 ) 

529 

530 # publishers can see preview of trending badge of their own snaps 

531 # on Publicise page 

532 show_as_preview = False 

533 if is_preview and authentication.is_authenticated(flask.session): 

534 show_as_preview = True 

535 

536 if context["trending"] or show_as_preview: 

537 svg = get_badge_svg( 

538 snap_name=snap_name, 

539 left_text=context["snap_title"], 

540 right_text="Trending this week", 

541 color="#FA7041", 

542 ) 

543 

544 return svg, 200, {"Content-Type": "image/svg+xml"} 

545 

546 @store.route('/install/<regex("' + snap_regex + '"):snap_name>/<distro>') 

547 def snap_distro_install(snap_name, distro): 

548 filename = f"store/content/distros/{distro}.yaml" 

549 distro_data = helpers.get_yaml(filename) 

550 

551 if not distro_data: 

552 flask.abort(404) 

553 

554 supported_archs = distro_data["supported-archs"] 

555 context = _get_context_snap_details(snap_name, supported_archs) 

556 

557 if all(arch not in context["channel_map"] for arch in supported_archs): 

558 return flask.render_template("404.html"), 404 

559 

560 context.update( 

561 { 

562 "distro": distro, 

563 "distro_name": distro_data["name"], 

564 "distro_logo": distro_data["logo"], 

565 "distro_logo_mono": distro_data["logo-mono"], 

566 "distro_color_1": distro_data["color-1"], 

567 "distro_color_2": distro_data["color-2"], 

568 "distro_install_steps": distro_data["install"], 

569 } 

570 ) 

571 cached_featured_snaps = redis_cache.get( 

572 "featured_snaps_install_pages", expected_type=list 

573 ) 

574 if cached_featured_snaps: 

575 context.update({"featured_snaps": cached_featured_snaps}) 

576 return flask.render_template( 

577 "store/snap-distro-install.html", **context 

578 ) 

579 try: 

580 featured_snaps_results = device_gateway.get_featured_items( 

581 size=13, page=1 

582 ).get("results", []) 

583 

584 except StoreApiError: 

585 featured_snaps_results = [] 

586 featured_snaps = [ 

587 snap 

588 for snap in featured_snaps_results 

589 if snap["package_name"] != snap_name 

590 ][:12] 

591 

592 for snap in featured_snaps: 

593 snap["icon_url"] = helpers.get_icon(snap["media"]) 

594 redis_cache.set( 

595 "featured_snaps_install_pages", featured_snaps, ttl=3600 

596 ) 

597 context.update({"featured_snaps": featured_snaps}) 

598 return flask.render_template( 

599 "store/snap-distro-install.html", **context 

600 ) 

601 

602 @store.route("/report", methods=["POST"]) 

603 def report_snap(): 

604 form_url = flask.current_app.config.get("REPORT_SHEET_URL") 

605 if not form_url: 

606 logger.warning("REPORT_SHEET_URL is not configured") 

607 return flask.jsonify({"error": "report_url_missing"}), 503 

608 

609 fields = flask.request.form 

610 

611 # If the honeypot is activated (hidden field populated 

612 # silently reject to avoid spam 

613 if "confirm" in fields: 

614 return flask.jsonify({"ok": True}), 200 

615 

616 if not verify_turnstile(fields.get("cf-turnstile-response", "")): 

617 return flask.jsonify({"error": "turnstile_failed"}), 400 

618 

619 payload = { 

620 "snap_name": fields.get("snap_name", ""), 

621 "reason": fields.get("reason", ""), 

622 "comment": fields.get("comment", ""), 

623 "email": fields.get("email", ""), 

624 } 

625 

626 try: 

627 response = requests.post(form_url, data=payload) 

628 if not response.ok: 

629 logger.warning( 

630 "Report sheet webhook returned %s", 

631 response.status_code, 

632 ) 

633 return flask.jsonify({"error": "report_failed"}), 502 

634 except requests.RequestException: 

635 logger.exception("Report sheet webhook request failed") 

636 return flask.jsonify({"error": "report_failed"}), 502 

637 

638 return flask.jsonify({"ok": True}), 200