Coverage for webapp/admin/views.py: 76%

1# Packages

2import os

3import json

4import flask

5from flask import make_response

6from canonicalwebteam.exceptions import (

7 StoreApiResponseErrorList,

8 StoreApiResourceNotFound,

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}

35def get_brand_id(session, store_id):

36 store = dashboard.get_store(session, store_id)

37 return store["brand-id"]

40@admin.route("/admin", defaults={"path": ""})

41@admin.route("/admin/<path:path>")

42@login_required

43@exchange_required

44def get_admin(path):

45 return flask.render_template("admin/admin.html", **context)

48@admin.route("/api/stores")

49@login_required

50@exchange_required

51def get_stores():

52 """

53 In this view we get all the stores the user is an admin or we show a 403

54 """

55 stores = dashboard.get_stores(flask.session)

57 res = {"success": True, "data": stores}

59 return jsonify(res)

62@admin.route("/api/store/<store_id>")

63@login_required

64@exchange_required

65def get_settings(store_id):

66 store = dashboard.get_store(flask.session, store_id)

67 store["links"] = []

69 if any(role["role"] == "admin" for role in store["roles"]):

70 store["links"].append(

71 {"name": "Members", "path": f'/admin/{store["id"]}/members'}

72 )

73 store["links"].append(

74 {"name": "Settings", "path": f'/admin/${store["id"]}/settings'}

75 )

77 res = {"success": True, "data": store}

79 return jsonify(res)

82@admin.route("/api/store/<store_id>/settings", methods=["PUT"])

83@login_required

84@exchange_required

85def post_settings(store_id):

86 settings = {}

87 settings["private"] = json.loads(flask.request.form.get("private"))

88 settings["manual-review-policy"] = flask.request.form.get(

89 "manual-review-policy"

90 )

92 res = {}

94 dashboard.change_store_settings(flask.session, store_id, settings)

95 res["msg"] = "Changes saved"

97 return jsonify({"success": True})

100@admin.route("/api/<store_id>/snaps/search")

101@login_required

102@exchange_required

103def get_snaps_search(store_id):

104 snaps = dashboard.get_store_snaps(

105 flask.session,

106 store_id,

107 flask.request.args.get("q"),

108 flask.request.args.get("allowed_for_inclusion"),

109 )

110

111 return jsonify(snaps)

112

113

114@admin.route("/api/store/<store_id>/snaps")

115@login_required

116@exchange_required

117def get_store_snaps(store_id):

118 snaps = dashboard.get_store_snaps(flask.session, store_id)

119 store = dashboard.get_store(flask.session, store_id)

120 if "store-whitelist" in store:

121 included_stores = []

122 for item in store["store-whitelist"]:

123 try:

124 store_item = dashboard.get_store(flask.session, item)

125 if store_item:

126 included_stores.append(

127 {

128 "id": store_item["id"],

129 "name": store_item["name"],

130 "userHasAccess": True,

131 }

132 )

133 except Exception:

134 included_stores.append(

135 {

136 "id": item,

137 "name": "Private store",

138 "userHasAccess": False,

139 }

140 )

141

142 if included_stores:

143 snaps.append({"included-stores": included_stores})

144 return jsonify(snaps)

145

146

147@admin.route("/api/store/<store_id>/snaps", methods=["POST"])

148@login_required

149@exchange_required

150def post_manage_store_snaps(store_id):

151 snaps = json.loads(flask.request.form.get("snaps"))

152

153 res = {}

154

155 dashboard.update_store_snaps(flask.session, store_id, snaps)

156 res["msg"] = "Changes saved"

157

158 return jsonify({"success": True})

159

160

161@admin.route("/api/store/<store_id>/members")

162@login_required

163@exchange_required

164def get_manage_members(store_id):

165 members = dashboard.get_store_members(flask.session, store_id)

166

167 for item in members:

168 if item["email"] == flask.session["publisher"]["email"]:

169 item["current_user"] = True

170

171 return jsonify(members)

172

173

174@admin.route("/api/store/<store_id>/members", methods=["POST"])

175@login_required

176@exchange_required

177def post_manage_members(store_id):

178 members = json.loads(flask.request.form.get("members"))

179

180 res = {}

181

182 try:

183 dashboard.update_store_members(flask.session, store_id, members)

184 res["msg"] = "Changes saved"

185 except StoreApiResponseErrorList as api_response_error_list:

186 codes = [error.get("code") for error in api_response_error_list.errors]

187

188 msgs = [

189 f"{error.get('message', 'An error occurred')}"

190 for error in api_response_error_list.errors

191 ]

192

193 for code in codes:

194 account_id = ""

195

196 if code == "store-users-no-match":

197 if account_id:

198 res["msg"] = code

199 else:

200 res["msg"] = "invite"

201

202 elif code == "store-users-multiple-matches":

203 res["msg"] = code

204 else:

205 for msg in msgs:

206 flask.flash(msg, "negative")

207

208 return jsonify(res)

209

210

211@admin.route("/api/store/<store_id>/invites")

212@login_required

213@exchange_required

214def get_invites(store_id):

215 invites = dashboard.get_store_invites(flask.session, store_id)

216

217 return jsonify(invites)

218

219

220@admin.route("/api/store/<store_id>/invite", methods=["POST"])

221@login_required

222@exchange_required

223def post_invite_members(store_id):

224 members = json.loads(flask.request.form.get("members"))

225

226 res = {}

227

228 try:

229 dashboard.invite_store_members(flask.session, store_id, members)

230 res["msg"] = "Changes saved"

231 except StoreApiResponseErrorList as api_response_error_list:

232 msgs = [

233 f"{error.get('message', 'An error occurred')}"

234 for error in api_response_error_list.errors

235 ]

236

237 msgs = list(dict.fromkeys(msgs))

238

239 for msg in msgs:

240 flask.flash(msg, "negative")

241

242 return jsonify(res)

243

244

245@admin.route("/api/store/<store_id>/invite/update", methods=["POST"])

246@login_required

247@exchange_required

248def update_invite_status(store_id):

249 invites = json.loads(flask.request.form.get("invites"))

250

251 res = {}

252

253 try:

254 dashboard.update_store_invites(flask.session, store_id, invites)

255 res["msg"] = "Changes saved"

256 except StoreApiResponseErrorList as api_response_error_list:

257 msgs = [

258 f"{error.get('message', 'An error occurred')}"

259 for error in api_response_error_list.errors

260 ]

261

262 msgs = list(dict.fromkeys(msgs))

263

264 for msg in msgs:

265 flask.flash(msg, "negative")

266

267 return jsonify(res)

268

269

270# ---------------------- MODELS SERVICES ----------------------

271@admin.route("/api/store/<store_id>/models")

272@login_required

273@exchange_required

274def get_models(store_id):

275 """

276 Retrieves models associated with a given store ID.

277

278 Args:

279 store_id (int): The ID of the store for which to retrieve models.

280

281 Returns:

282 dict: A dictionary containing the response message, success status,

283 and data.

284 """

285 res = {}

286 try:

287 models = publisher_gateway.get_store_models(flask.session, store_id)

288 res["success"] = True

289 res["data"] = models

290 response = make_response(res, 200)

291 response.cache_control.max_age = "3600"

292 except StoreApiResponseErrorList as error_list:

293 error_messages = [

294 f"{error.get('message', 'An error occurred')}"

295 for error in error_list.errors

296 ]

297 if "unauthorized" in error_messages:

298 res["message"] = "Store not found"

299 else:

300 res["message"] = " ".join(error_messages)

301 res["success"] = False

302 response = make_response(res, 500)

303

304 return response

305

306

307@admin.route("/api/store/<store_id>/models", methods=["POST"])

308@login_required

309@exchange_required

310def create_models(store_id: str):

311 """

312 Create a model for a given store.

313

314 Args:

315 store_id (str): The ID of the store.

316

317 Returns:

318 dict: A dictionary containing the response message and success

319 status.

320 """

321

322 # TO DO: Addn validation that name does not exist already

323

324 res = {}

325

326 try:

327 name = flask.request.form.get("name")

328 api_key = flask.request.form.get("api_key", "")

329

330 if len(name) > 128:

331 res["message"] = "Name is too long. Limit 128 characters"

332 res["success"] = False

333 return make_response(res, 500)

334

335 if api_key and len(api_key) != 50 and not api_key.isalpha():

336 res["message"] = "Invalid API key"

337 res["success"] = False

338 return make_response(res, 500)

339

340 publisher_gateway.create_store_model(

341 flask.session, store_id, name, api_key

342 )

343 res["success"] = True

344

345 return make_response(res, 201)

346 except StoreApiResponseErrorList as error_list:

347 res["success"] = False

348 messages = [

349 f"{error.get('message', 'An error occurred')}"

350 for error in error_list.errors

351 ]

352 res["message"] = (" ").join(messages)

353

354 except Exception:

355 res["success"] = False

356 res["message"] = "An error occurred"

357

358 return make_response(res, 500)

359

360

361@admin.route("/api/store/<store_id>/models/<model_name>", methods=["PATCH"])

362@login_required

363@exchange_required

364def update_model(store_id: str, model_name: str):

365 """

366 Update a model for a given store.

367

368 Args:

369 store_id (str): The ID of the store.

370 model_name (str): The name of the model.

371

372 Returns:

373 dict: A dictionary containing the response message and success

374 status.

375 """

376 res = {}

377

378 try:

379 api_key = flask.request.form.get("api_key", "")

380

381 if len(api_key) != 50 and not api_key.isalpha():

382 res["message"] = "Invalid API key"

383 res["success"] = False

384 return make_response(res, 500)

385

386 publisher_gateway.update_store_model(

387 flask.session, store_id, model_name, api_key

388 )

389 res["success"] = True

390

391 except StoreApiResponseErrorList as error_list:

392 res["success"] = False

393 res["message"] = error_list.errors[0]["message"]

394

395 except StoreApiResourceNotFound:

396 res["success"] = False

397 res["message"] = "Model not found"

398 if res["success"]:

399 return make_response(res, 200)

400 return make_response(res, 500)

401

402

403@admin.route("/api/store/<store_id>/models/<model_name>/policies")

404@login_required

405@exchange_required

406def get_policies(store_id: str, model_name: str):

407 """

408 Get the policies for a given store model.

409

410 Args:

411 store_id (str): The ID of the store.

412 model_name (str): The name of the model.

413

414 Returns:

415 dict: A dictionary containing the response message and success

416 """

417 brand_id = get_brand_id(flask.session, store_id)

418 res = {}

419

420 try:

421 policies = publisher_gateway.get_store_model_policies(

422 flask.session, brand_id, model_name

423 )

424 res["success"] = True

425 res["data"] = policies

426 response = make_response(res, 200)

427 response.cache_control.max_age = "3600"

428 return response

429 except StoreApiResponseErrorList as error_list:

430 res["success"] = False

431 res["message"] = " ".join(

432 [

433 f"{error.get('message', 'An error occurred')}"

434 for error in error_list.errors

435 ]

436 )

437 except Exception:

438 res["success"] = False

439 res["message"] = "An error occurred"

440

441 return make_response(res, 500)

442

443

444@admin.route(

445 "/api/store/<store_id>/models/<model_name>/policies", methods=["POST"]

446)

447@login_required

448@exchange_required

449def create_policy(store_id: str, model_name: str):

450 """

451 Creat policy for a store model.

452

453 Args:

454 store_id (str): The ID of the store.

455 model_name (str): The name of the model.

456

457 Returns:

458 dict: A dictionary containing the response message and success

459 """

460 signing_key = flask.request.form.get("signing_key")

461 res = {}

462 try:

463 signing_keys_data = publisher_gateway.get_store_signing_keys(

464 flask.session, store_id

465 )

466 signing_keys = [key["sha3-384"] for key in signing_keys_data]

467

468 if not signing_key:

469 res["message"] = "Signing key required"

470 res["success"] = False

471 return make_response(res, 500)

472

473 if signing_key in signing_keys:

474 publisher_gateway.create_store_model_policy(

475 flask.session, store_id, model_name, signing_key

476 )

477 res["success"] = True

478 else:

479 res["message"] = "Invalid signing key"

480 res["success"] = False

481 except StoreApiResponseErrorList as error_list:

482 res["success"] = False

483 res["message"] = error_list.errors[0]["message"]

484

485 if res["success"]:

486 return make_response(res, 200)

487 return make_response(res, 500)

488

489

490@admin.route(

491 "/api/store/<store_id>/models/<model_name>/policies/<revision>",

492 methods=["DELETE"],

493)

494@login_required

495@exchange_required

496def delete_policy(store_id: str, model_name: str, revision: str):

497 res = {}

498 try:

499 response = publisher_gateway.delete_store_model_policy(

500 flask.session, store_id, model_name, revision

501 )

502 if response.status_code == 204:

503 res = {"success": True}

504 if response.status_code == 404:

505 res = {"success": False, "message": "Policy not found"}

506 except StoreApiResponseErrorList as error_list:

507 res["success"] = False

508 res["message"] = error_list.errors[0]["message"]

509 if res["success"]:

510 return make_response(res, 200)

511 return make_response(res, 500)

512

513

514@admin.route("/api/store/<store_id>/brand")

515@login_required

516@exchange_required

517def get_brand_store(store_id: str):

518 brand_id = get_brand_id(flask.session, store_id)

519 res = {}

520 try:

521 brand = publisher_gateway.get_brand(flask.session, brand_id)

522

523 res["data"] = brand

524 res["success"] = True

525

526 except StoreApiResponseErrorList as error_list:

527 res["success"] = False

528 res["message"] = " ".join(

529 [

530 f"{error.get('message', 'An error occurred')}"

531 for error in error_list.errors

532 ]

533 )

534 res["data"] = []

535

536 response = make_response(res)

537 response.cache_control.max_age = 3600

538

539 return response

540

541

542@admin.route("/api/store/<store_id>/signing-keys")

543@login_required

544@exchange_required

545def get_signing_keys(store_id: str):

546 brand_id = get_brand_id(flask.session, store_id)

547 res = {}

548 try:

549 signing_keys = publisher_gateway.get_store_signing_keys(

550 flask.session, brand_id

551 )

552 res["data"] = signing_keys

553 res["success"] = True

554 response = make_response(res, 200)

555 response.cache_control.max_age = 3600

556 return response

557 except StoreApiResponseErrorList as error_list:

558 res["success"] = False

559 res["success"] = False

560 res["message"] = " ".join(

561 [

562 f"{error.get('message', 'An error occurred')}"

563 for error in error_list.errors

564 ]

565 )

566 res["data"] = []

567 return make_response(res, 500)

568

569

570@admin.route("/api/store/<store_id>/signing-keys", methods=["POST"])

571@login_required

572@exchange_required

573def create_signing_key(store_id: str):

574 name = flask.request.form.get("name")

575 res = {}

576

577 try:

578 if name and len(name) <= 128:

579 publisher_gateway.create_store_signing_key(

580 flask.session, store_id, name

581 )

582 res["success"] = True

583 return make_response(res, 200)

584 else:

585 res["message"] = "Invalid signing key. Limit 128 characters"

586 res["success"] = False

587 make_response(res, 500)

588 except StoreApiResponseErrorList as error_list:

589 res["success"] = False

590 res["message"] = error_list.errors[0]["message"]

591

592 return make_response(res, 500)

593

594

595@admin.route(

596 "/api/store/<store_id>/signing-keys/<signing_key_sha3_384>",

597 methods=["DELETE"],

598)

599@login_required

600@exchange_required

601def delete_signing_key(store_id: str, signing_key_sha3_384: str):

602 """

603 Deletes a signing key from the store.

604

605 Args:

606 store_id (str): The ID of the store.

607 signing_key_sha3_384 (str): The signing key to delete.

608

609 Returns:

610 Response: A response object with the following fields:

611 - success (bool): True if the signing key was deleted successfully,

612 False otherwise.

613 - message (str): A message describing the result of the deletion.

614 - data (dict): A dictionary containing models where the signing

615 key is used.

616 """

617 res = {}

618

619 try:

620 response = publisher_gateway.delete_store_signing_key(

621 flask.session, store_id, signing_key_sha3_384

622 )

623

624 if response.status_code == 204:

625 res["success"] = True

626 return make_response(res, 200)

627 elif response.status_code == 404:

628 res["success"] = False

629 res["message"] = "Signing key not found"

630 return make_response(res, 404)

631 except StoreApiResponseErrorList as error_list:

632 message = error_list.errors[0]["message"]

633 if (

634 error_list.status_code == 409

635 and "used to sign at least one serial policy" in message

636 ):

637 matching_models = []

638 models_response = get_models(store_id).json

639 models = models_response.get("data", [])

640

641 for model in models:

642 policies_resp = get_policies(store_id, model["name"]).json

643 policies = policies_resp.get("data", [])

644 matching_policies = [

645 {"revision": policy["revision"]}

646 for policy in policies

647 if policy["signing-key-sha3-384"] == signing_key_sha3_384

648 ]

649 if matching_policies:

650 matching_models.append(

651 {

652 "name": model["name"],

653 "policies": matching_policies,

654 }

655 )

656 res["data"] = {"models": matching_models}

657 res["message"] = "Signing key is used in at least one policy"

658 res["success"] = False

659 else:

660 res["success"] = False

661 res["message"] = error_list.errors[0]["message"]

662

663 return make_response(res, 500)

664

665

666# ---------------------- END MODELS SERVICES ----------------------

667

668

669# -------------------- FEATURED SNAPS AUTOMATION ------------------

670@admin.route("/admin/featured", methods=["POST"])

671@login_required

672@exchange_required

673def post_featured_snaps():

674 """

675 In this view, we do three things:

676 1. Fetch all currently featured snaps

677 2. Delete the currently featured snaps

678 3. Update featured snaps to be newly featured

679

680 Args:

681 None

682

683 Returns:

684 dict: A dictionary containing the response message and success status.

685 """

686

687 # new_featured_snaps is the list of featured snaps to be updated

688 featured_snaps = flask.request.form.get("snaps")

689

690 if not featured_snaps:

691 response = {

692 "success": False,

693 "message": "Snaps cannot be empty",

694 }

695 return make_response(response, 500)

696 new_featured_snaps = featured_snaps.split(",")

697

698 # currently_featured_snap is the list of featured snaps to be deleted

699 currently_featured_snaps = []

700

701 next = True

702 while next:

703 featured_snaps = device_gateway.get_featured_snaps()

704 currently_featured_snaps.extend(

705 featured_snaps.get("_embedded", {}).get("clickindex:package", [])

706 )

707 next = featured_snaps.get("_links", {}).get("next", False)

708

709 currently_featured_snap_ids = [

710 snap["snap_id"] for snap in currently_featured_snaps

711 ]

712

713 delete_response = publisher_gateway.delete_featured_snaps(

714 flask.session, {"packages": currently_featured_snap_ids}

715 )

716 if delete_response.status_code != 201:

717 response = {

718 "success": False,

719 "message": "An error occurred while deleting featured snaps",

720 }

721 return make_response(response, 500)

722 snap_ids = [

723 dashboard.get_snap_id(flask.session, snap_name)

724 for snap_name in new_featured_snaps

725 ]

726

727 update_response = publisher_gateway.update_featured_snaps(

728 flask.session, snap_ids

729 )

730 if update_response.status_code != 201:

731 response = {

732 "success": False,

733 "message": "An error occured while updating featured snaps",

734 }

735 return make_response(response, 500)

736 return make_response({"success": True}, 200)