Coverage for webapp/store/snap_details

1import flask

2from flask import Response

3import requests

5import logging

6import humanize

7import os

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

17from canonicalwebteam.flask_base.decorators import (

18 exclude_xframe_options_header,

19)

20from canonicalwebteam.exceptions import StoreApiError

21from canonicalwebteam.store_api.devicegw import DeviceGW

22from pybadges import badge

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

25device_gateway_sbom = DeviceGW("sbom", helpers.api_session)

27logger = logging.getLogger(__name__)

30FIELDS = [

31 "title",

32 "summary",

33 "description",

34 "license",

35 "contact",

36 "website",

37 "publisher",

38 "media",

39 "download",

40 "version",

41 "created-at",

42 "confinement",

43 "categories",

44 "trending",

45 "unlisted",

46 "links",

47 "revision",

48 "sboms",

49]

51FIELDS_EXTRA_DETAILS = [

52 "aliases",

53]

56def snap_details_views(store):

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

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

60 def _get_context_snap_details(snap_name, supported_architectures=None):

61 details = device_gateway.get_item_details(

62 snap_name, fields=FIELDS, api_version=2

63 )

64 # 404 for any snap under quarantine

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

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

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

69 # responds that the snaps still exists with data.

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

71 # For example: mir-kiosk-browser

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

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

75 formatted_description = parse_markdown_description(

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

77 )

79 channel_maps_list = logic.convert_channel_maps(

80 details.get("channel-map")

81 )

83 latest_channel = logic.get_last_updated_version(

84 details.get("channel-map")

85 )

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

89 default_track = (

90 details.get("default-track")

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

92 else "latest"

93 )

95 lowest_risk_available = logic.get_lowest_available_risk(

96 channel_maps_list, default_track

97 )

99 extracted_info = logic.extract_info_channel_map(

100 channel_maps_list, default_track, lowest_risk_available

101 )

102

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

104 updates = logic.get_latest_versions(

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

106 default_track,

107 lowest_risk_available,

108 supported_architectures,

109 )

110

111 # Determine the most recent update date from updates tuple

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

113 # recent non-stable

114 most_recent_update = None

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

116 # Compare both and use the most recent

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

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

119 if date_0 and date_1:

120 most_recent_update = max(date_0, date_1)

121 else:

122 most_recent_update = date_0 or date_1

123 elif updates[0]:

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

125 elif updates[1]:

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

127

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

129

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

131 screenshots = logic.filter_screenshots(

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

133 )

134

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

136

137 publisher_info = helpers.get_yaml(

138 "{}{}.yaml".format(

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

140 "PUBLISHER_PAGES"

141 ],

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

143 ),

144 typ="safe",

145 )

146

147 publisher_snaps = []

148 publisher_featured_snaps = None

149

150 if publisher_info:

151 publisher_results = logic.get_publisher_snaps(

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

153 )

154

155 snaps_by_name = {}

156 for snap in publisher_results:

157 item = snap["snap"]

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

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

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

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

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

163 }

164

165 publisher_featured_snaps = logic.hydrate_featured_snaps(

166 publisher_info.get("featured_snaps"), snaps_by_name

167 )

168

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

170 # the snap currently being viewed.

171 excluded_names = {snap_name}

172 for snap in publisher_featured_snaps:

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

174

175 available_snaps = [

176 snap

177 for name, snap in snaps_by_name.items()

178 if name not in excluded_names

179 ]

180

181 publisher_snaps = logic.get_n_random_snaps(available_snaps, 4)

182

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

184

185 is_users_snap = False

186 if authentication.is_authenticated(flask.session):

187 if (

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

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

190 ):

191 is_users_snap = True

192

193 # build list of categories of a snap

194 categories = logic.get_snap_categories(

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

196 )

197

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

199

200 is_last_updated_old = logic.is_snap_old(last_updated)

201

202 context = {

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

204 # Data direct from details API

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

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

207 "categories": categories,

208 "icon_url": icon_url,

209 "version": extracted_info["version"],

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

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

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

213 "screenshots": screenshots,

214 "video": video,

215 "publisher_snaps": publisher_snaps,

216 "publisher_featured_snaps": publisher_featured_snaps,

217 "has_publisher_page": publisher_info is not None,

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

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

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

221 "description": formatted_description,

222 "channel_map": channel_maps_list,

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

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

225 "default_track": default_track,

226 "lowest_risk_available": lowest_risk_available,

227 "confinement": extracted_info["confinement"],

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

229 # Transformed API data

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

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

232 "last_updated_raw": last_updated,

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

234 "is_last_updated_old": is_last_updated_old,

235 "is_users_snap": is_users_snap,

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

237 "developer": developer,

238 # TODO: This is horrible and hacky

239 "appliances": {

240 "adguard-home": "adguard",

241 "mosquitto": "mosquitto",

242 "nextcloud": "nextcloud",

243 "plexmediaserver": "plex",

244 "openhab": "openhab",

245 },

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

247 "updates": updates,

248 "revisions": revisions,

249 "turnstile_site_key": (

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

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

252 else ""

253 ),

254 }

255 return context

256

257 def verify_turnstile(turnstile_response):

258 turnstile_secret = flask.current_app.config.get(

259 "TURNSTILE_SECRET_KEY", ""

260 )

261 if not turnstile_secret:

262 return True

263

264 if not turnstile_response:

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

266 return False

267

268 payload = {

269 "secret": turnstile_secret,

270 "response": turnstile_response,

271 }

272

273 try:

274 response = requests.post(

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

276 data=payload,

277 timeout=10,

278 )

279 if not response.ok:

280 logger.warning(

281 "Turnstile verification returned %s",

282 response.status_code,

283 )

284 return False

285 verification = response.json()

286 except (requests.RequestException, ValueError):

287 logger.exception("Turnstile verification failed")

288 return False

289

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

291 logger.warning(

292 "Turnstile verification denied report: %s",

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

294 )

295 return False

296

297 return True

298

299 def snap_has_sboms(revisions, snap_id):

300 if not revisions:

301 return False

302

303 sbom_path = f"download/sbom_snap_{snap_id}_{revisions[0]}.spdx2.3.json"

304 endpoint = device_gateway_sbom.get_endpoint_url(sbom_path)

305

306 res = requests.head(endpoint)

307

308 # backend returns 302 instead of 200 for a successful request

309 # adding the check for 200 in case this is changed without us knowing

310 if res.status_code == 200 or res.status_code == 302:

311 return True

312

313 return False

314

315 @store.route("/download/sbom_snap_<snap_id>_<revision>.spdx2.3.json")

316 def get_sbom(snap_id, revision):

317 sbom_path = f"download/sbom_snap_{snap_id}_{revision}.spdx2.3.json"

318 endpoint = device_gateway_sbom.get_endpoint_url(sbom_path)

319

320 res = requests.get(endpoint)

321

322 return flask.jsonify(res.json())

323

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

325 def snap_details(snap_name):

326 """

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

328

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

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

331 with appropriate sanitation.

332 """

333

334 error_info = {}

335 status_code = 200

336

337 context = _get_context_snap_details(snap_name)

338 try:

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

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

341 # snaps that don't use that channel

342 extra_details = device_gateway.get_snap_details(

343 snap_name, channel="", fields=FIELDS_EXTRA_DETAILS

344 )

345 except Exception:

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

347 extra_details = None

348

349 if extra_details and extra_details["aliases"]:

350 context["aliases"] = [

351 [

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

353 alias_obj["name"],

354 ]

355 for alias_obj in extra_details["aliases"]

356 ]

357

358 country_metric_name = "weekly_installed_base_by_country_percent"

359 os_metric_name = "weekly_installed_base_by_operating_system_normalized"

360

361 end = metrics_helper.get_last_metrics_processed_date()

362

363 metrics_query_json = [

364 metrics_helper.get_filter(

365 metric_name=country_metric_name,

366 snap_id=context["snap_id"],

367 start=end,

368 end=end,

369 ),

370 metrics_helper.get_filter(

371 metric_name=os_metric_name,

372 snap_id=context["snap_id"],

373 start=end,

374 end=end,

375 ),

376 ]

377

378 metrics_response = device_gateway.get_public_metrics(

379 metrics_query_json

380 )

381

382 os_metrics = None

383 country_devices = None

384 if metrics_response:

385 oses = metrics_helper.find_metric(metrics_response, os_metric_name)

386 os_metrics = metrics.OsMetric(

387 name=oses["metric_name"],

388 series=oses["series"],

389 buckets=oses["buckets"],

390 status=oses["status"],

391 )

392

393 territories = metrics_helper.find_metric(

394 metrics_response, country_metric_name

395 )

396 country_devices = metrics.CountryDevices(

397 name=territories["metric_name"],

398 series=territories["series"],

399 buckets=territories["buckets"],

400 status=territories["status"],

401 private=False,

402 )

403

404 has_sboms = snap_has_sboms(context["revisions"], context["snap_id"])

405

406 context.update(

407 {

408 "countries": (

409 country_devices.country_data if country_devices else None

410 ),

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

412 # Context info

413 "is_linux": (

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

415 and "Android"

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

417 ),

418 "error_info": error_info,

419 }

420 )

421

422 context["has_sboms"] = has_sboms

423

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

425 context["channel_map"].keys()

426 )

427

428 return (

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

430 status_code,

431 )

432

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

434 @exclude_xframe_options_header

435 def snap_details_embedded(snap_name):

436 """

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

438

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

440 some of the data through to the template,

441 with appropriate sanitation.

442 """

443 status_code = 200

444

445 context = _get_context_snap_details(snap_name)

446

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

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

449 if button and button not in button_variants:

450 button = "black"

451

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

453

454 context.update(

455 {

456 "default_architecture": (

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

458 ),

459 "button": button,

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

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

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

463 }

464 )

465

466 return (

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

468 status_code,

469 )

470

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

472 def snap_details_case_sensitive(snap_name):

473 return flask.redirect(

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

475 )

476

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

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

479 snap_link = flask.request.url_root + snap_name

480

481 svg = badge(

482 left_text=left_text if show_name else "",

483 right_text=right_text,

484 right_color=color,

485 left_link=snap_link,

486 right_link=snap_link,

487 logo=(

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

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

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

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

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

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

494 ),

495 )

496 return svg

497

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

499 def snap_details_badge(snap_name):

500 context = _get_context_snap_details(snap_name)

501

502 # channel with safest risk available in default track

503 snap_channel = "".join(

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

505 )

506

507 svg = get_badge_svg(

508 snap_name=snap_name,

509 left_text=context["snap_title"],

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

511 )

512

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

514

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

516 def snap_install_badge(lang, theme):

517 base_path = "static/images/badges/"

518 allowed_langs = helpers.list_folders(base_path)

519

520 if lang not in allowed_langs:

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

522

523 file_name = (

524 "snap-store-white.svg"

525 if theme == "light"

526 else "snap-store-black.svg"

527 )

528

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

530

531 # Ensure the path is within the base path

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

533 return Response(

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

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

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

537 mimetype="image/svg+xml",

538 status=404,

539 )

540 else:

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

542 svg_content = svg_file.read()

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

544

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

546 def snap_details_badge_trending(snap_name):

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

548 context = _get_context_snap_details(snap_name)

549

550 # default to empty SVG

551 svg = (

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

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

554 )

555

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

557 # on Publicise page

558 show_as_preview = False

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

560 show_as_preview = True

561

562 if context["trending"] or show_as_preview:

563 svg = get_badge_svg(

564 snap_name=snap_name,

565 left_text=context["snap_title"],

566 right_text="Trending this week",

567 color="#FA7041",

568 )

569

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

571

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

573 def snap_distro_install(snap_name, distro):

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

575 distro_data = helpers.get_yaml(filename)

576

577 if not distro_data:

578 flask.abort(404)

579

580 supported_archs = distro_data["supported-archs"]

581 context = _get_context_snap_details(snap_name, supported_archs)

582

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

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

585

586 context.update(

587 {

588 "distro": distro,

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

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

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

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

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

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

595 }

596 )

597 cached_featured_snaps = redis_cache.get(

598 "featured_snaps_install_pages", expected_type=list

599 )

600 if cached_featured_snaps:

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

602 return flask.render_template(

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

604 )

605 try:

606 featured_snaps_results = device_gateway.get_featured_items(

607 size=13, page=1

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

609

610 except StoreApiError:

611 featured_snaps_results = []

612 featured_snaps = [

613 snap

614 for snap in featured_snaps_results

615 if snap["package_name"] != snap_name

616 ][:12]

617

618 for snap in featured_snaps:

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

620 redis_cache.set(

621 "featured_snaps_install_pages", featured_snaps, ttl=3600

622 )

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

624 return flask.render_template(

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

626 )

627

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

629 def report_snap():

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

631 if not form_url:

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

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

634

635 fields = flask.request.form

636

637 # If the honeypot is activated (hidden field populated

638 # silently reject to avoid spam

639 if "confirm" in fields:

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

641

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

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

644

645 payload = {

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

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

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

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

650 }

651

652 try:

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

654 if not response.ok:

655 logger.warning(

656 "Report sheet webhook returned %s",

657 response.status_code,

658 )

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

660 except requests.RequestException:

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

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

663

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

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

240 statements