Coverage for webapp/admin/views.py: 75%
391 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-28 22:05 +0000
1# Packages
2import os
3import json
4import flask
5from flask import make_response
6from canonicalwebteam.exceptions import (
7 StoreApiResponseErrorList,
8 StoreApiResourceNotFound,
9)
10from canonicalwebteam.store_api.dashboard import Dashboard
11from canonicalwebteam.store_api.publishergw import PublisherGW
12from canonicalwebteam.store_api.devicegw import DeviceGW
13from flask.json import jsonify
15# Local
16from webapp.decorators import login_required, exchange_required
17from webapp.helpers import api_publisher_session, api_session
20dashboard = Dashboard(api_session)
21publisher_gateway = PublisherGW("snap", api_publisher_session)
22device_gateway = DeviceGW("snap", api_session)
24admin = flask.Blueprint(
25 "admin", __name__, template_folder="/templates", static_folder="/static"
26)
28SNAPSTORE_DASHBOARD_API_URL = os.getenv(
29 "SNAPSTORE_DASHBOARD_API_URL", "https://dashboard.snapcraft.io/"
30)
32context = {"api_url": SNAPSTORE_DASHBOARD_API_URL}
35@admin.route("/admin", defaults={"path": ""})
36@admin.route("/admin/<path:path>")
37@login_required
38@exchange_required
39def get_admin(path):
40 return flask.render_template("admin/admin.html", **context)
43@admin.route("/api/stores")
44@login_required
45@exchange_required
46def get_stores():
47 """
48 In this view we get all the stores the user is an admin or we show a 403
49 """
50 stores = dashboard.get_stores(flask.session)
52 res = {"success": True, "data": stores}
54 return jsonify(res)
57@admin.route("/api/store/<store_id>")
58@login_required
59@exchange_required
60def get_settings(store_id):
61 store = dashboard.get_store(flask.session, store_id)
62 store["links"] = []
64 if any(role["role"] == "admin" for role in store["roles"]):
65 store["links"].append(
66 {"name": "Members", "path": f'/admin/{store["id"]}/members'}
67 )
68 store["links"].append(
69 {"name": "Settings", "path": f'/admin/${store["id"]}/settings'}
70 )
72 res = {"success": True, "data": store}
74 return jsonify(res)
77@admin.route("/api/store/<store_id>/settings", methods=["PUT"])
78@login_required
79@exchange_required
80def post_settings(store_id):
81 settings = {}
82 settings["private"] = json.loads(flask.request.form.get("private"))
83 settings["manual-review-policy"] = flask.request.form.get(
84 "manual-review-policy"
85 )
87 res = {}
89 dashboard.change_store_settings(flask.session, store_id, settings)
90 res["msg"] = "Changes saved"
92 return jsonify({"success": True})
95@admin.route("/api/<store_id>/snaps/search")
96@login_required
97@exchange_required
98def get_snaps_search(store_id):
99 snaps = dashboard.get_store_snaps(
100 flask.session,
101 store_id,
102 flask.request.args.get("q"),
103 flask.request.args.get("allowed_for_inclusion"),
104 )
106 return jsonify(snaps)
109@admin.route("/api/store/<store_id>/snaps")
110@login_required
111@exchange_required
112def get_store_snaps(store_id):
113 snaps = dashboard.get_store_snaps(flask.session, store_id)
114 store = dashboard.get_store(flask.session, store_id)
115 if "store-whitelist" in store:
116 included_stores = []
117 for item in store["store-whitelist"]:
118 try:
119 store_item = dashboard.get_store(flask.session, item)
120 if store_item:
121 included_stores.append(
122 {
123 "id": store_item["id"],
124 "name": store_item["name"],
125 "userHasAccess": True,
126 }
127 )
128 except Exception:
129 included_stores.append(
130 {
131 "id": item,
132 "name": "Private store",
133 "userHasAccess": False,
134 }
135 )
137 if included_stores:
138 snaps.append({"included-stores": included_stores})
139 return jsonify(snaps)
142@admin.route("/api/store/<store_id>/snaps", methods=["POST"])
143@login_required
144@exchange_required
145def post_manage_store_snaps(store_id):
146 snaps = json.loads(flask.request.form.get("snaps"))
148 res = {}
150 dashboard.update_store_snaps(flask.session, store_id, snaps)
151 res["msg"] = "Changes saved"
153 return jsonify({"success": True})
156@admin.route("/api/store/<store_id>/members")
157@login_required
158@exchange_required
159def get_manage_members(store_id):
160 members = dashboard.get_store_members(flask.session, store_id)
162 for item in members:
163 if item["email"] == flask.session["publisher"]["email"]:
164 item["current_user"] = True
166 return jsonify(members)
169@admin.route("/api/store/<store_id>/members", methods=["POST"])
170@login_required
171@exchange_required
172def post_manage_members(store_id):
173 members = json.loads(flask.request.form.get("members"))
175 res = {}
177 try:
178 dashboard.update_store_members(flask.session, store_id, members)
179 res["msg"] = "Changes saved"
180 except StoreApiResponseErrorList as api_response_error_list:
181 codes = [error.get("code") for error in api_response_error_list.errors]
183 msgs = [
184 f"{error.get('message', 'An error occurred')}"
185 for error in api_response_error_list.errors
186 ]
188 for code in codes:
189 account_id = ""
191 if code == "store-users-no-match":
192 if account_id:
193 res["msg"] = code
194 else:
195 res["msg"] = "invite"
197 elif code == "store-users-multiple-matches":
198 res["msg"] = code
199 else:
200 for msg in msgs:
201 flask.flash(msg, "negative")
203 return jsonify(res)
206@admin.route("/api/store/<store_id>/invites")
207@login_required
208@exchange_required
209def get_invites(store_id):
210 invites = dashboard.get_store_invites(flask.session, store_id)
212 return jsonify(invites)
215@admin.route("/api/store/<store_id>/invite", methods=["POST"])
216@login_required
217@exchange_required
218def post_invite_members(store_id):
219 members = json.loads(flask.request.form.get("members"))
221 res = {}
223 try:
224 dashboard.invite_store_members(flask.session, store_id, members)
225 res["msg"] = "Changes saved"
226 except StoreApiResponseErrorList as api_response_error_list:
227 msgs = [
228 f"{error.get('message', 'An error occurred')}"
229 for error in api_response_error_list.errors
230 ]
232 msgs = list(dict.fromkeys(msgs))
234 for msg in msgs:
235 flask.flash(msg, "negative")
237 return jsonify(res)
240@admin.route("/api/store/<store_id>/invite/update", methods=["POST"])
241@login_required
242@exchange_required
243def update_invite_status(store_id):
244 invites = json.loads(flask.request.form.get("invites"))
246 res = {}
248 try:
249 dashboard.update_store_invites(flask.session, store_id, invites)
250 res["msg"] = "Changes saved"
251 except StoreApiResponseErrorList as api_response_error_list:
252 msgs = [
253 f"{error.get('message', 'An error occurred')}"
254 for error in api_response_error_list.errors
255 ]
257 msgs = list(dict.fromkeys(msgs))
259 for msg in msgs:
260 flask.flash(msg, "negative")
262 return jsonify(res)
265# ---------------------- MODELS SERVICES ----------------------
266@admin.route("/api/store/<store_id>/models")
267@login_required
268@exchange_required
269def get_models(store_id):
270 """
271 Retrieves models associated with a given store ID.
273 Args:
274 store_id (int): The ID of the store for which to retrieve models.
276 Returns:
277 dict: A dictionary containing the response message, success status,
278 and data.
279 """
280 res = {}
281 try:
282 models = publisher_gateway.get_store_models(flask.session, store_id)
283 res["success"] = True
284 res["data"] = models
285 response = make_response(res, 200)
286 response.cache_control.max_age = "3600"
287 except StoreApiResponseErrorList as error_list:
288 error_messages = [
289 f"{error.get('message', 'An error occurred')}"
290 for error in error_list.errors
291 ]
292 if "unauthorized" in error_messages:
293 res["message"] = "Store not found"
294 else:
295 res["message"] = " ".join(error_messages)
296 res["success"] = False
297 response = make_response(res, 500)
299 return response
302@admin.route("/api/store/<store_id>/models", methods=["POST"])
303@login_required
304@exchange_required
305def create_models(store_id: str):
306 """
307 Create a model for a given store.
309 Args:
310 store_id (str): The ID of the store.
312 Returns:
313 dict: A dictionary containing the response message and success
314 status.
315 """
317 # TO DO: Addn validation that name does not exist already
319 res = {}
321 try:
322 name = flask.request.form.get("name")
323 api_key = flask.request.form.get("api_key", "")
325 if len(name) > 128:
326 res["message"] = "Name is too long. Limit 128 characters"
327 res["success"] = False
328 return make_response(res, 500)
330 if api_key and len(api_key) != 50 and not api_key.isalpha():
331 res["message"] = "Invalid API key"
332 res["success"] = False
333 return make_response(res, 500)
335 publisher_gateway.create_store_model(
336 flask.session, store_id, name, api_key
337 )
338 res["success"] = True
340 return make_response(res, 201)
341 except StoreApiResponseErrorList as error_list:
342 res["success"] = False
343 messages = [
344 f"{error.get('message', 'An error occurred')}"
345 for error in error_list.errors
346 ]
347 res["message"] = (" ").join(messages)
349 except Exception:
350 res["success"] = False
351 res["message"] = "An error occurred"
353 return make_response(res, 500)
356@admin.route("/api/store/<store_id>/models/<model_name>", methods=["PATCH"])
357@login_required
358@exchange_required
359def update_model(store_id: str, model_name: str):
360 """
361 Update a model for a given store.
363 Args:
364 store_id (str): The ID of the store.
365 model_name (str): The name of the model.
367 Returns:
368 dict: A dictionary containing the response message and success
369 status.
370 """
371 res = {}
373 try:
374 api_key = flask.request.form.get("api_key", "")
376 if len(api_key) != 50 and not api_key.isalpha():
377 res["message"] = "Invalid API key"
378 res["success"] = False
379 return make_response(res, 500)
381 publisher_gateway.update_store_model(
382 flask.session, store_id, model_name, api_key
383 )
384 res["success"] = True
386 except StoreApiResponseErrorList as error_list:
387 res["success"] = False
388 res["message"] = error_list.errors[0]["message"]
390 except StoreApiResourceNotFound:
391 res["success"] = False
392 res["message"] = "Model not found"
393 if res["success"]:
394 return make_response(res, 200)
395 return make_response(res, 500)
398@admin.route("/api/store/<store_id>/models/<model_name>/policies")
399@login_required
400@exchange_required
401def get_policies(store_id: str, model_name: str):
402 """
403 Get the policies for a given store model.
405 Args:
406 store_id (str): The ID of the store.
407 model_name (str): The name of the model.
409 Returns:
410 dict: A dictionary containing the response message and success
411 """
412 res = {}
414 try:
415 policies = publisher_gateway.get_store_model_policies(
416 flask.session, store_id, model_name
417 )
418 res["success"] = True
419 res["data"] = policies
420 response = make_response(res, 200)
421 response.cache_control.max_age = "3600"
422 return response
423 except StoreApiResponseErrorList as error_list:
424 res["success"] = False
425 res["message"] = " ".join(
426 [
427 f"{error.get('message', 'An error occurred')}"
428 for error in error_list.errors
429 ]
430 )
431 except Exception:
432 res["success"] = False
433 res["message"] = "An error occurred"
435 return make_response(res, 500)
438@admin.route(
439 "/api/store/<store_id>/models/<model_name>/policies", methods=["POST"]
440)
441@login_required
442@exchange_required
443def create_policy(store_id: str, model_name: str):
444 """
445 Creat policy for a store model.
447 Args:
448 store_id (str): The ID of the store.
449 model_name (str): The name of the model.
451 Returns:
452 dict: A dictionary containing the response message and success
453 """
454 signing_key = flask.request.form.get("signing_key")
455 res = {}
456 try:
457 signing_keys_data = publisher_gateway.get_store_signing_keys(
458 flask.session, store_id
459 )
460 signing_keys = [key["sha3-384"] for key in signing_keys_data]
462 if not signing_key:
463 res["message"] = "Signing key required"
464 res["success"] = False
465 return make_response(res, 500)
467 if signing_key in signing_keys:
468 publisher_gateway.create_store_model_policy(
469 flask.session, store_id, model_name, signing_key
470 )
471 res["success"] = True
472 else:
473 res["message"] = "Invalid signing key"
474 res["success"] = False
475 except StoreApiResponseErrorList as error_list:
476 res["success"] = False
477 res["message"] = error_list.errors[0]["message"]
479 if res["success"]:
480 return make_response(res, 200)
481 return make_response(res, 500)
484@admin.route(
485 "/api/store/<store_id>/models/<model_name>/policies/<revision>",
486 methods=["DELETE"],
487)
488@login_required
489@exchange_required
490def delete_policy(store_id: str, model_name: str, revision: str):
491 res = {}
492 try:
493 response = publisher_gateway.delete_store_model_policy(
494 flask.session, store_id, model_name, revision
495 )
496 if response.status_code == 204:
497 res = {"success": True}
498 if response.status_code == 404:
499 res = {"success": False, "message": "Policy not found"}
500 except StoreApiResponseErrorList as error_list:
501 res["success"] = False
502 res["message"] = error_list.errors[0]["message"]
503 if res["success"]:
504 return make_response(res, 200)
505 return make_response(res, 500)
508@admin.route("/api/store/<store_id>/brand")
509@login_required
510@exchange_required
511def get_brand_store(store_id: str):
512 res = {}
513 try:
514 brand = publisher_gateway.get_brand(flask.session, store_id)
516 res["data"] = brand
517 res["success"] = True
519 except StoreApiResponseErrorList as error_list:
520 res["success"] = False
521 res["message"] = " ".join(
522 [
523 f"{error.get('message', 'An error occurred')}"
524 for error in error_list.errors
525 ]
526 )
527 res["data"] = []
529 response = make_response(res)
530 response.cache_control.max_age = 3600
532 return response
535@admin.route("/api/store/<store_id>/signing-keys")
536@login_required
537@exchange_required
538def get_signing_keys(store_id: str):
539 res = {}
540 try:
541 signing_keys = publisher_gateway.get_store_signing_keys(
542 flask.session, store_id
543 )
544 res["data"] = signing_keys
545 res["success"] = True
546 response = make_response(res, 200)
547 response.cache_control.max_age = 3600
548 return response
549 except StoreApiResponseErrorList as error_list:
550 res["success"] = False
551 res["success"] = False
552 res["message"] = " ".join(
553 [
554 f"{error.get('message', 'An error occurred')}"
555 for error in error_list.errors
556 ]
557 )
558 res["data"] = []
559 return make_response(res, 500)
562@admin.route("/api/store/<store_id>/signing-keys", methods=["POST"])
563@login_required
564@exchange_required
565def create_signing_key(store_id: str):
566 name = flask.request.form.get("name")
567 res = {}
569 try:
570 if name and len(name) <= 128:
571 publisher_gateway.create_store_signing_key(
572 flask.session, store_id, name
573 )
574 res["success"] = True
575 return make_response(res, 200)
576 else:
577 res["message"] = "Invalid signing key. Limit 128 characters"
578 res["success"] = False
579 make_response(res, 500)
580 except StoreApiResponseErrorList as error_list:
581 res["success"] = False
582 res["message"] = error_list.errors[0]["message"]
584 return make_response(res, 500)
587@admin.route(
588 "/api/store/<store_id>/signing-keys/<signing_key_sha3_384>",
589 methods=["DELETE"],
590)
591@login_required
592@exchange_required
593def delete_signing_key(store_id: str, signing_key_sha3_384: str):
594 """
595 Deletes a signing key from the store.
597 Args:
598 store_id (str): The ID of the store.
599 signing_key_sha3_384 (str): The signing key to delete.
601 Returns:
602 Response: A response object with the following fields:
603 - success (bool): True if the signing key was deleted successfully,
604 False otherwise.
605 - message (str): A message describing the result of the deletion.
606 - data (dict): A dictionary containing models where the signing
607 key is used.
608 """
609 res = {}
611 try:
612 response = publisher_gateway.delete_store_signing_key(
613 flask.session, store_id, signing_key_sha3_384
614 )
616 if response.status_code == 204:
617 res["success"] = True
618 return make_response(res, 200)
619 elif response.status_code == 404:
620 res["success"] = False
621 res["message"] = "Signing key not found"
622 return make_response(res, 404)
623 except StoreApiResponseErrorList as error_list:
624 message = error_list.errors[0]["message"]
625 if (
626 error_list.status_code == 409
627 and "used to sign at least one serial policy" in message
628 ):
629 matching_models = []
630 models_response = get_models(store_id).json
631 models = models_response.get("data", [])
633 for model in models:
634 policies_resp = get_policies(store_id, model["name"]).json
635 policies = policies_resp.get("data", [])
636 matching_policies = [
637 {"revision": policy["revision"]}
638 for policy in policies
639 if policy["signing-key-sha3-384"] == signing_key_sha3_384
640 ]
641 if matching_policies:
642 matching_models.append(
643 {
644 "name": model["name"],
645 "policies": matching_policies,
646 }
647 )
648 res["data"] = {"models": matching_models}
649 res["message"] = "Signing key is used in at least one policy"
650 res["success"] = False
651 else:
652 res["success"] = False
653 res["message"] = error_list.errors[0]["message"]
655 return make_response(res, 500)
658# ---------------------- END MODELS SERVICES ----------------------
661# -------------------- FEATURED SNAPS AUTOMATION ------------------
662@admin.route("/admin/featured", methods=["POST"])
663@login_required
664@exchange_required
665def post_featured_snaps():
666 """
667 In this view, we do three things:
668 1. Fetch all currently featured snaps
669 2. Delete the currently featured snaps
670 3. Update featured snaps to be newly featured
672 Args:
673 None
675 Returns:
676 dict: A dictionary containing the response message and success status.
677 """
679 # new_featured_snaps is the list of featured snaps to be updated
680 featured_snaps = flask.request.form.get("snaps")
682 if not featured_snaps:
683 response = {
684 "success": False,
685 "message": "Snaps cannot be empty",
686 }
687 return make_response(response, 500)
688 new_featured_snaps = featured_snaps.split(",")
690 # currently_featured_snap is the list of featured snaps to be deleted
691 currently_featured_snaps = []
693 next = True
694 while next:
695 featured_snaps = device_gateway.get_featured_snaps()
696 currently_featured_snaps.extend(
697 featured_snaps.get("_embedded", {}).get("clickindex:package", [])
698 )
699 next = featured_snaps.get("_links", {}).get("next", False)
701 currently_featured_snap_ids = [
702 snap["snap_id"] for snap in currently_featured_snaps
703 ]
705 delete_response = publisher_gateway.delete_featured_snaps(
706 flask.session, {"packages": currently_featured_snap_ids}
707 )
708 if delete_response.status_code != 201:
709 response = {
710 "success": False,
711 "message": "An error occurred while deleting featured snaps",
712 }
713 return make_response(response, 500)
714 snap_ids = [
715 dashboard.get_snap_id(flask.session, snap_name)
716 for snap_name in new_featured_snaps
717 ]
719 update_response = publisher_gateway.update_featured_snaps(
720 flask.session, snap_ids
721 )
722 if update_response.status_code != 201:
723 response = {
724 "success": False,
725 "message": "An error occured while updating featured snaps",
726 }
727 return make_response(response, 500)
728 return make_response({"success": True}, 200)