Coverage for webapp / endpoints / snaps.py: 65%

1import flask 

2from flask import make_response 

3from flask.json import jsonify 

4import json 

5 

6import dns.resolver 

7import re 

8 

9import webapp.helpers as helpers 

10from webapp.decorators import login_required, exchange_required 

11 

12from canonicalwebteam.store_api.devicegw import DeviceGW 

13from canonicalwebteam.store_api.dashboard import Dashboard 

14 

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

16dashboard = Dashboard(helpers.api_session) 

17 

18FIELDS = [ 

19 "title", 

20 "summary", 

21 "description", 

22 "license", 

23 "contact", 

24 "website", 

25 "publisher", 

26 "media", 

27 "download", 

28 "version", 

29 "created-at", 

30 "confinement", 

31 "categories", 

32 "trending", 

33 "unlisted", 

34 "links", 

35] 

36snaps = flask.Blueprint( 

37 "snaps", 

38 __name__, 

39) 

40 

41 

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

43 

44 

45def _get_snap_link_fields(snap_name): 

46 details = device_gateway.get_item_details( 

47 snap_name, api_version=2, fields=FIELDS 

48 ) 

49 return { 

50 "links": details["snap"].get("links", {}), 

51 } 

52 

53 

54@snaps.route('/api/<regex("' + snap_regex + '"):snap_name>/verify') 

55def dns_verified_status(snap_name): 

56 res = {"primary_domain": False, "token": None} 

57 context = _get_snap_link_fields(snap_name) 

58 

59 primary_domain = None 

60 

61 if "website" in context["links"]: 

62 primary_domain = context["links"]["website"][0] 

63 

64 if primary_domain: 

65 token = helpers.get_dns_verification_token(snap_name, primary_domain) 

66 

67 domain = re.compile(r"https?://(www\.)?") 

68 domain = domain.sub("", primary_domain).strip().strip("/") 

69 

70 res["token"] = token 

71 

72 try: 

73 dns_txt_records = [ 

74 dns_record.to_text() 

75 for dns_record in dns.resolver.resolve(domain, "TXT").rrset 

76 ] 

77 

78 if f'"SNAPCRAFT_IO_VERIFICATION={token}"' in dns_txt_records: 

79 res["primary_domain"] = True 

80 

81 except Exception: 

82 res["primary_domain"] = False 

83 

84 response = make_response(res, 200) 

85 response.cache_control.max_age = "3600" 

86 return response 

87 

88 

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

90@login_required 

91@exchange_required 

92def get_store_snaps(store_id): 

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

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

95 if "store-whitelist" in store: 

96 included_stores = [] 

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

98 try: 

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

100 if store_item: 

101 included_stores.append( 

102 { 

103 "id": store_item["id"], 

104 "name": store_item["name"], 

105 "userHasAccess": True, 

106 } 

107 ) 

108 except Exception: 

109 included_stores.append( 

110 { 

111 "id": item, 

112 "name": "Private store", 

113 "userHasAccess": False, 

114 } 

115 ) 

116 

117 if included_stores: 

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

119 return jsonify(snaps) 

120 

121 

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

123@login_required 

124@exchange_required 

125def post_manage_store_snaps(store_id): 

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

127 

128 res = {} 

129 

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

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

132 

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