1# Packages 

2import os 

3import json 

4import flask 

5from flask import make_response 

6from canonicalwebteam.store_api.exceptions import ( 

7 StoreApiResponseErrorList, 

8 StoreApiResourceNotFound, 

9) 

10from canonicalwebteam.store_api.stores.snapstore import ( 

11 SnapStoreAdmin, 

12 SnapPublisher, 

13) 

14from flask.json import jsonify 

15 

16# Local 

17from webapp.decorators import login_required, exchange_required 

18from webapp.helpers import api_publisher_session 

19 

20admin_api = SnapStoreAdmin(api_publisher_session) 

21publisher_api = SnapPublisher(api_publisher_session) 

22 

23admin = flask.Blueprint( 

24 "admin", __name__, template_folder="/templates", static_folder="/static" 

25) 

26 

27SNAPSTORE_DASHBOARD_API_URL = os.getenv( 

28 "SNAPSTORE_DASHBOARD_API_URL", "https://dashboard.snapcraft.io/" 

29) 

30 

31context = {"api_url": SNAPSTORE_DASHBOARD_API_URL} 

32 

33 

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

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

36@login_required 

37@exchange_required 

38def get_admin(path): 

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

40 

41 

42@admin.route("/admin/stores") 

43@login_required 

44@exchange_required 

45def get_stores(): 

46 """ 

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

48 """ 

49 stores = admin_api.get_stores(flask.session) 

50 

51 return jsonify(stores) 

52 

53 

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

55@login_required 

56@exchange_required 

57def get_settings(store_id): 

58 store = admin_api.get_store(flask.session, store_id) 

59 

60 return jsonify(store) 

61 

62 

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

64@login_required 

65@exchange_required 

66def post_settings(store_id): 

67 settings = {} 

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

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

70 "manual-review-policy" 

71 ) 

72 

73 res = {} 

74 

75 admin_api.change_store_settings(flask.session, store_id, settings) 

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

77 

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

79 

80 

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

82@login_required 

83@exchange_required 

84def get_snaps_search(store_id): 

85 snaps = admin_api.get_store_snaps( 

86 flask.session, 

87 store_id, 

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

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

90 ) 

91 

92 return jsonify(snaps) 

93 

94 

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

96@login_required 

97@exchange_required 

98def get_store_snaps(store_id): 

99 snaps = admin_api.get_store_snaps(flask.session, store_id) 

100 store = admin_api.get_store(flask.session, store_id) 

101 if "store-whitelist" in store: 

102 included_stores = [] 

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

104 try: 

105 store_item = admin_api.get_store(flask.session, item) 

106 if store_item: 

107 included_stores.append( 

108 { 

109 "id": store_item["id"], 

110 "name": store_item["name"], 

111 "userHasAccess": True, 

112 } 

113 ) 

114 except Exception: 

115 included_stores.append( 

116 { 

117 "id": item, 

118 "name": "Private store", 

119 "userHasAccess": False, 

120 } 

121 ) 

122 

123 if included_stores: 

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

125 return jsonify(snaps) 

126 

127 

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

129@login_required 

130@exchange_required 

131def post_manage_store_snaps(store_id): 

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

133 

134 res = {} 

135 

136 admin_api.update_store_snaps(flask.session, store_id, snaps) 

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

138 

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

140 

141 

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

143@login_required 

144@exchange_required 

145def get_manage_members(store_id): 

146 members = admin_api.get_store_members(flask.session, store_id) 

147 

148 for item in members: 

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

150 item["current_user"] = True 

151 

152 return jsonify(members) 

153 

154 

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

156@login_required 

157@exchange_required 

158def post_manage_members(store_id): 

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

160 

161 res = {} 

162 

163 try: 

164 admin_api.update_store_members(flask.session, store_id, members) 

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

166 except StoreApiResponseErrorList as api_response_error_list: 

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

168 

169 msgs = [ 

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

171 for error in api_response_error_list.errors 

172 ] 

173 

174 for code in codes: 

175 account_id = "" 

176 

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

178 if account_id: 

179 res["msg"] = code 

180 else: 

181 res["msg"] = "invite" 

182 

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

184 res["msg"] = code 

185 else: 

186 for msg in msgs: 

187 flask.flash(msg, "negative") 

188 

189 return jsonify(res) 

190 

191 

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

193@login_required 

194@exchange_required 

195def get_invites(store_id): 

196 invites = admin_api.get_store_invites(flask.session, store_id) 

197 

198 return jsonify(invites) 

199 

200 

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

202@login_required 

203@exchange_required 

204def post_invite_members(store_id): 

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

206 

207 res = {} 

208 

209 try: 

210 admin_api.invite_store_members(flask.session, store_id, members) 

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

212 except StoreApiResponseErrorList as api_response_error_list: 

213 msgs = [ 

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

215 for error in api_response_error_list.errors 

216 ] 

217 

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

219 

220 for msg in msgs: 

221 flask.flash(msg, "negative") 

222 

223 return jsonify(res) 

224 

225 

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

227@login_required 

228@exchange_required 

229def update_invite_status(store_id): 

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

231 

232 res = {} 

233 

234 try: 

235 admin_api.update_store_invites(flask.session, store_id, invites) 

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

237 except StoreApiResponseErrorList as api_response_error_list: 

238 msgs = [ 

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

240 for error in api_response_error_list.errors 

241 ] 

242 

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

244 

245 for msg in msgs: 

246 flask.flash(msg, "negative") 

247 

248 return jsonify(res) 

249 

250 

251# ---------------------- MODELS SERVICES ---------------------- 

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

253@login_required 

254@exchange_required 

255def get_models(store_id): 

256 """ 

257 Retrieves models associated with a given store ID. 

258 

259 Args: 

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

261 

262 Returns: 

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

264 and data. 

265 """ 

266 res = {} 

267 try: 

268 models = admin_api.get_store_models(flask.session, store_id) 

269 res["success"] = True 

270 res["data"] = models 

271 response = make_response(res, 200) 

272 response.cache_control.max_age = "3600" 

273 except StoreApiResponseErrorList as error_list: 

274 error_messages = [ 

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

276 for error in error_list.errors 

277 ] 

278 if "unauthorized" in error_messages: 

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

280 else: 

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

282 res["success"] = False 

283 response = make_response(res, 500) 

284 

285 return response 

286 

287 

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

289@login_required 

290@exchange_required 

291def create_models(store_id: str): 

292 """ 

293 Create a model for a given store. 

294 

295 Args: 

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

297 

298 Returns: 

299 dict: A dictionary containing the response message and success 

300 status. 

301 """ 

302 

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

304 

305 res = {} 

306 

307 try: 

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

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

310 

311 if len(name) > 128: 

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

313 res["success"] = False 

314 return make_response(res, 500) 

315 

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

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

318 res["success"] = False 

319 return make_response(res, 500) 

320 

321 admin_api.create_store_model(flask.session, store_id, name, api_key) 

322 res["success"] = True 

323 

324 return make_response(res, 201) 

325 except StoreApiResponseErrorList as error_list: 

326 res["success"] = False 

327 messages = [ 

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

329 for error in error_list.errors 

330 ] 

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

332 

333 except Exception: 

334 res["success"] = False 

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

336 

337 return make_response(res, 500) 

338 

339 

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

341@login_required 

342@exchange_required 

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

344 """ 

345 Update a model for a given store. 

346 

347 Args: 

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

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

350 

351 Returns: 

352 dict: A dictionary containing the response message and success 

353 status. 

354 """ 

355 res = {} 

356 

357 try: 

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

359 

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

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

362 res["success"] = False 

363 return make_response(res, 500) 

364 

365 admin_api.update_store_model( 

366 flask.session, store_id, model_name, api_key 

367 ) 

368 res["success"] = True 

369 

370 except StoreApiResponseErrorList as error_list: 

371 res["success"] = False 

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

373 

374 except StoreApiResourceNotFound: 

375 res["success"] = False 

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

377 if res["success"]: 

378 return make_response(res, 200) 

379 return make_response(res, 500) 

380 

381 

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

383@login_required 

384@exchange_required 

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

386 """ 

387 Get the policies for a given store model. 

388 

389 Args: 

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

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

392 

393 Returns: 

394 dict: A dictionary containing the response message and success 

395 """ 

396 res = {} 

397 

398 try: 

399 policies = admin_api.get_store_model_policies( 

400 flask.session, store_id, model_name 

401 ) 

402 res["success"] = True 

403 res["data"] = policies 

404 response = make_response(res, 200) 

405 response.cache_control.max_age = "3600" 

406 return response 

407 except StoreApiResponseErrorList as error_list: 

408 res["success"] = False 

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

410 [ 

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

412 for error in error_list.errors 

413 ] 

414 ) 

415 except Exception: 

416 res["success"] = False 

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

418 

419 return make_response(res, 500) 

420 

421 

422@admin.route( 

423 "/admin/store/<store_id>/models/<model_name>/policies", methods=["POST"] 

424) 

425@login_required 

426@exchange_required 

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

428 """ 

429 Creat policy for a store model. 

430 

431 Args: 

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

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

434 

435 Returns: 

436 dict: A dictionary containing the response message and success 

437 """ 

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

439 res = {} 

440 try: 

441 signing_keys_data = admin_api.get_store_signing_keys( 

442 flask.session, store_id 

443 ) 

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

445 

446 if not signing_key: 

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

448 res["success"] = False 

449 return make_response(res, 500) 

450 

451 if signing_key in signing_keys: 

452 admin_api.create_store_model_policy( 

453 flask.session, store_id, model_name, signing_key 

454 ) 

455 res["success"] = True 

456 else: 

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

458 res["success"] = False 

459 except StoreApiResponseErrorList as error_list: 

460 res["success"] = False 

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

462 

463 if res["success"]: 

464 return make_response(res, 200) 

465 return make_response(res, 500) 

466 

467 

468@admin.route( 

469 "/admin/store/<store_id>/models/<model_name>/policies/<revision>", 

470 methods=["DELETE"], 

471) 

472@login_required 

473@exchange_required 

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

475 res = {} 

476 try: 

477 response = admin_api.delete_store_model_policy( 

478 flask.session, store_id, model_name, revision 

479 ) 

480 if response.status_code == 204: 

481 res = {"success": True} 

482 if response.status_code == 404: 

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

484 except StoreApiResponseErrorList as error_list: 

485 res["success"] = False 

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

487 if res["success"]: 

488 return make_response(res, 200) 

489 return make_response(res, 500) 

490 

491 

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

493@login_required 

494@exchange_required 

495def get_brand_store(store_id: str): 

496 res = {} 

497 try: 

498 brand = admin_api.get_brand(flask.session, store_id) 

499 

500 res["data"] = brand 

501 res["success"] = True 

502 

503 except StoreApiResponseErrorList as error_list: 

504 res["success"] = False 

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

506 [ 

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

508 for error in error_list.errors 

509 ] 

510 ) 

511 res["data"] = [] 

512 

513 response = make_response(res) 

514 response.cache_control.max_age = 3600 

515 

516 return response 

517 

518 

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

520@login_required 

521@exchange_required 

522def get_signing_keys(store_id: str): 

523 res = {} 

524 try: 

525 signing_keys = admin_api.get_store_signing_keys( 

526 flask.session, store_id 

527 ) 

528 res["data"] = signing_keys 

529 res["success"] = True 

530 response = make_response(res, 200) 

531 response.cache_control.max_age = 3600 

532 return response 

533 except StoreApiResponseErrorList as error_list: 

534 res["success"] = False 

535 res["success"] = False 

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

537 [ 

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

539 for error in error_list.errors 

540 ] 

541 ) 

542 res["data"] = [] 

543 return make_response(res, 500) 

544 

545 

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

547@login_required 

548@exchange_required 

549def create_signing_key(store_id: str): 

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

551 res = {} 

552 

553 try: 

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

555 admin_api.create_store_signing_key(flask.session, store_id, name) 

556 res["success"] = True 

557 return make_response(res, 200) 

558 else: 

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

560 res["success"] = False 

561 make_response(res, 500) 

562 except StoreApiResponseErrorList as error_list: 

563 res["success"] = False 

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

565 

566 return make_response(res, 500) 

567 

568 

569@admin.route( 

570 "/admin/store/<store_id>/signing-keys/<signing_key_sha3_384>", 

571 methods=["DELETE"], 

572) 

573@login_required 

574@exchange_required 

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

576 """ 

577 Deletes a signing key from the store. 

578 

579 Args: 

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

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

582 

583 Returns: 

584 Response: A response object with the following fields: 

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

586 False otherwise. 

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

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

589 key is used. 

590 """ 

591 res = {} 

592 

593 try: 

594 response = admin_api.delete_store_signing_key( 

595 flask.session, store_id, signing_key_sha3_384 

596 ) 

597 

598 if response.status_code == 204: 

599 res["success"] = True 

600 return make_response(res, 200) 

601 elif response.status_code == 404: 

602 res["success"] = False 

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

604 return make_response(res, 404) 

605 except StoreApiResponseErrorList as error_list: 

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

607 if ( 

608 error_list.status_code == 409 

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

610 ): 

611 matching_models = [] 

612 models_response = get_models(store_id).json 

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

614 

615 for model in models: 

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

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

618 matching_policies = [ 

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

620 for policy in policies 

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

622 ] 

623 if matching_policies: 

624 matching_models.append( 

625 { 

626 "name": model["name"], 

627 "policies": matching_policies, 

628 } 

629 ) 

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

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

632 res["success"] = False 

633 else: 

634 res["success"] = False 

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

636 

637 return make_response(res, 500) 

638 

639 

640# ---------------------- END MODELS SERVICES ---------------------- 

641 

642 

643# -------------------- FEATURED SNAPS AUTOMATION ------------------ 

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

645@login_required 

646@exchange_required 

647def post_featured_snaps(): 

648 """ 

649 In this view, we do three things: 

650 1. Fetch all currently featured snaps 

651 2. Delete the currently featured snaps 

652 3. Update featured snaps to be newly featured 

653 

654 Args: 

655 None 

656 

657 Returns: 

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

659 """ 

660 

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

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

663 

664 if not featured_snaps: 

665 response = { 

666 "success": False, 

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

668 } 

669 return make_response(response, 500) 

670 new_featured_snaps = featured_snaps.split(",") 

671 

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

673 currently_featured_snaps = [] 

674 

675 next = True 

676 while next: 

677 featured_snaps = admin_api.get_featured_snaps(flask.session) 

678 currently_featured_snaps.extend( 

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

680 ) 

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

682 

683 currently_featured_snap_ids = [ 

684 snap["snap_id"] for snap in currently_featured_snaps 

685 ] 

686 

687 delete_response = admin_api.delete_featured_snaps( 

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

689 ) 

690 if delete_response.status_code != 201: 

691 response = { 

692 "success": False, 

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

694 } 

695 return make_response(response, 500) 

696 snap_ids = [ 

697 publisher_api.get_snap_id(snap_name, flask.session) 

698 for snap_name in new_featured_snaps 

699 ] 

700 

701 update_response = admin_api.update_featured_snaps( 

702 flask.session, {"packages": snap_ids} 

703 ) 

704 if update_response.status_code != 201: 

705 response = { 

706 "success": False, 

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

708 } 

709 return make_response(response, 500) 

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