Coverage for webapp/helpers.py: 84%

1import json 

2import os 

3import hashlib 

4import sentry_sdk 

5 

6import flask 

7from canonicalwebteam.launchpad import Launchpad 

8from ruamel.yaml import YAML 

9from webapp.api.requests import PublisherSession, Session 

10from canonicalwebteam.store_api.dashboard import Dashboard 

11import webapp.api.marketo as marketo_api 

12 

13_yaml = YAML(typ="rt") 

14_yaml_safe = YAML(typ="safe") 

15api_session = Session() 

16api_publisher_session = PublisherSession() 

17marketo = marketo_api.Marketo() 

18dashboard = Dashboard(api_session) 

19 

20launchpad = Launchpad( 

21 username=os.getenv("LP_API_USERNAME"), 

22 token=os.getenv("LP_API_TOKEN"), 

23 secret=os.getenv("LP_API_TOKEN_SECRET"), 

24 session=api_publisher_session, 

25) 

26 

27 

28def get_yaml_loader(typ="safe"): 

29 if typ == "safe": 

30 return _yaml_safe 

31 return _yaml 

32 

33 

34def get_licenses(): 

35 try: 

36 with open("webapp/licenses.json") as f: 

37 licenses = json.load(f)["licenses"] 

38 

39 def _build_custom_license(license_id, license_name): 

40 return {"licenseId": license_id, "name": license_name} 

41 

42 CUSTOM_LICENSES = [ 

43 _build_custom_license("Proprietary", "Proprietary"), 

44 _build_custom_license("Other Open Source", "Other Open Source"), 

45 _build_custom_license( 

46 "AGPL-3.0+", "GNU Affero General Public License v3.0 or later" 

47 ), 

48 ] 

49 

50 licenses = licenses + CUSTOM_LICENSES 

51 except Exception: 

52 licenses = [] 

53 

54 return licenses 

55 

56 

57def is_valid_path(path): 

58 base_path = os.path.abspath(flask.current_app.root_path) 

59 target_path = os.path.abspath(os.path.join(base_path, path)) 

60 return target_path.startswith(base_path) 

61 

62 

63def get_file(filename, replaces={}): 

64 """ 

65 Reads a file, replaces occurences of all the keys in `replaces` with 

66 the correspondant values and returns the resulting string or None 

67 

68 Keyword arguments: 

69 filename -- name if the file to load. 

70 replaces -- key/values to replace in the file content (default {}) 

71 """ 

72 if not is_valid_path(filename): 

73 return None 

74 

75 filepath = os.path.join(flask.current_app.root_path, filename) 

76 

77 try: 

78 with open(filepath, "r") as f: 

79 data = f.read() 

80 for key in replaces: 

81 data = data.replace(key, replaces[key]) 

82 except Exception: 

83 data = None 

84 

85 return data 

86 

87 

88def get_yaml(filename, typ="safe", replaces={}): 

89 """ 

90 Reads a file, replaces occurences of all the keys in `replaces` with the 

91 correspondant values and returns an ordered dict with the YAML content 

92 

93 Keyword arguments: 

94 filename -- name if the file to load. 

95 typ -- type of yaml loader 

96 replaces -- key/values to replace in the file content (default {}) 

97 """ 

98 try: 

99 yaml = get_yaml_loader(typ) 

100 data = get_file(filename, replaces) 

101 return yaml.load(data) 

102 except Exception: 

103 return None 

104 

105 

106def dump_yaml(data, stream, typ="safe"): 

107 yaml = get_yaml_loader(typ) 

108 yaml.dump(data, stream) 

109 

110 

111def get_icon(media): 

112 icons = [m["url"] for m in media if m["type"] == "icon"] 

113 if len(icons) > 0: 

114 return icons[0] 

115 return "" 

116 

117 

118def get_publisher_data(): 

119 # We don't use the data from this endpoint. 

120 # It is mostly used to make sure the user has signed 

121 # the terms and conditions. 

122 dashboard.get_account(flask.session) 

123 

124 flask_user = flask.session["publisher"] 

125 

126 subscriptions = None 

127 

128 # don't rely on marketo to show the page, 

129 # if anything fails, just continue and don't show 

130 # this section 

131 try: 

132 subscribed_to_newsletter = False 

133 marketo_user = marketo.get_user(flask_user["email"]) 

134 if marketo_user: 

135 marketo_subscribed = marketo.get_newsletter_subscription( 

136 marketo_user["id"] 

137 ) 

138 if marketo_subscribed.get("snapcraftnewsletter"): 

139 subscribed_to_newsletter = True 

140 

141 subscriptions = {"newsletter": subscribed_to_newsletter} 

142 except Exception: 

143 sentry_sdk.capture_exception() 

144 

145 flask_user["subscriptions"] = subscriptions 

146 context = {"publisher": flask_user} 

147 

148 return context 

149 

150 

151def get_dns_verification_token(snap_name, domain): 

152 salt = os.getenv("DNS_VERIFICATION_SALT") 

153 token_string = f"{domain}:{snap_name}:{salt}" 

154 token = hashlib.sha256(token_string.encode("utf-8")).hexdigest() 

155 return token 

156 

157 

158def get_csp_as_str(csp={}): 

159 csp_str = "" 

160 for key, values in csp.items(): 

161 csp_value = " ".join(values) 

162 csp_str += f"{key} {csp_value}; " 

163 return csp_str.strip() 

164 

165 

166def list_folders(directory): 

167 return [ 

168 item 

169 for item in os.listdir(directory) 

170 if os.path.isdir(os.path.join(directory, item)) 

171 ] 

172 

173 

174def directory_exists(file): 

175 if not is_valid_path(file): 

176 return False 

177 

178 target_path = os.path.abspath( 

179 os.path.join(flask.current_app.root_path, file) 

180 ) 

181 return os.path.isdir(target_path) 

182 

183 

184def get_brand_id(session, store_id): 

185 store = dashboard.get_store(session, store_id) 

186 return store["brand-id"]