Coverage for webapp/helpers.py: 66%

1import re 

2import json 

3from urllib.parse import urlparse, urlencode 

4 

5from bs4 import BeautifulSoup 

6from ruamel.yaml import YAML 

7from slugify import slugify 

8from datetime import datetime, timedelta 

9import mistune 

10import bleach 

11from canonicalwebteam.discourse import DiscourseAPI 

12from dateutil import parser 

13import requests 

14 

15session = requests.Session() 

16discourse_api = DiscourseAPI( 

17 base_url="https://discourse.charmhub.io/", 

18 session=session, 

19) 

20 

21_yaml = YAML(typ="rt") 

22_yaml_safe = YAML(typ="safe") 

23 

24 

25def get_yaml_loader(typ="safe"): 

26 if typ == "safe": 

27 return _yaml_safe 

28 return _yaml 

29 

30 

31def is_safe_url(url): 

32 """ 

33 Return True if the URL is inside the same app 

34 """ 

35 if not url: 

36 return False 

37 

38 cleaned_url = url.strip() 

39 

40 if cleaned_url.startswith(("/", "\\")) and not cleaned_url.startswith( 

41 ("//", "\\\\", "/\\", "\\/") 

42 ): 

43 parsed = urlparse(cleaned_url) 

44 return not parsed.scheme and not parsed.netloc 

45 

46 return False 

47 

48 

49def get_soup(html_content): 

50 soup = BeautifulSoup(html_content, "html.parser") 

51 return soup 

52 

53 

54# Change all the headers (if step=2: eg h1 => h3) 

55def decrease_header(header, step): 

56 level = int(header.name[1:]) + step 

57 if level > 6: 

58 level = 6 

59 header.name = f"h{str(level)}" 

60 

61 return header 

62 

63 

64def add_header_id(h, levels): 

65 id = slugify(h.get_text()) 

66 level = int(h.name[1:]) 

67 

68 # Go through previous headings and find any that are lower 

69 levels.append((level, id)) 

70 reversed_levels = list(reversed(levels)) 

71 parents = [] 

72 level_cache = None 

73 for i in reversed_levels: 

74 if i[0] < level and not level_cache: 

75 parents.append(i) 

76 level_cache = i[0] 

77 elif i[0] < level and i[0] < level_cache: 

78 parents.append(i) 

79 level_cache = i[0] 

80 parents.reverse() 

81 if "id" not in h.attrs: 

82 parent_path_id = "" 

83 if len(parents) > 0: 

84 parent_path_id = "--".join([i[1] for i in parents]) + "--" 

85 h["id"] = parent_path_id + id 

86 

87 return h 

88 

89 

90def modify_headers(soup, decrease_step=2): 

91 levels = [] 

92 

93 for header in soup.find_all(re.compile("^h[1-6]$")): 

94 decrease_header(header, decrease_step) 

95 add_header_id(header, levels) 

96 

97 return soup 

98 

99 

100def schedule_banner(start_date: str, end_date: str): 

101 try: 

102 end = datetime.strptime(end_date, "%Y-%m-%d") 

103 start = datetime.strptime(start_date, "%Y-%m-%d") 

104 present = datetime.now() 

105 return start <= present < end 

106 except ValueError: 

107 return False 

108 

109 

110def markdown_to_html(markdown_text): 

111 markdown = mistune.create_markdown(renderer=mistune.HTMLRenderer()) 

112 return markdown(markdown_text) 

113 

114 

115ALLOWED_HTML_TAGS = set(bleach.sanitizer.ALLOWED_TAGS).union( 

116 { 

117 "br", 

118 "code", 

119 "div", 

120 "h1", 

121 "h2", 

122 "h3", 

123 "h4", 

124 "h5", 

125 "h6", 

126 "hr", 

127 "img", 

128 "p", 

129 "pre", 

130 "span", 

131 "table", 

132 "tbody", 

133 "td", 

134 "tfoot", 

135 "th", 

136 "thead", 

137 "tr", 

138 } 

139) 

140ALLOWED_HTML_ATTRIBUTES = { 

141 **bleach.sanitizer.ALLOWED_ATTRIBUTES, 

142 "*": ["class", "id"], 

143 "a": ["href", "rel", "target", "title"], 

144 "img": ["alt", "height", "src", "title", "width"], 

145 "td": ["colspan", "rowspan"], 

146 "th": ["colspan", "rowspan"], 

147} 

148ALLOWED_HTML_PROTOCOLS = ["http", "https", "mailto", "tel"] 

149 

150 

151def sanitize_html(html_content): 

152 if not html_content: 

153 return "" 

154 return bleach.clean( 

155 html_content, 

156 tags=ALLOWED_HTML_TAGS, 

157 attributes=ALLOWED_HTML_ATTRIBUTES, 

158 protocols=ALLOWED_HTML_PROTOCOLS, 

159 strip=True, 

160 ) 

161 

162 

163def param_redirect_capture(req, resp): 

164 """ 

165 Functions that captures params and sets a cookie based on a match 

166 with a predefined list. 

167 """ 

168 # Signatures to capture in a cookie 

169 param_signatures = [ 

170 {"endpoint": "/accept-invite", "params": ["package", "token"]} 

171 ] 

172 path = req.path 

173 params = req.args 

174 

175 for item in param_signatures: 

176 # If the endpoint and all required params are present 

177 if item["endpoint"] == path and set(item["params"]).issubset( 

178 set(params) 

179 ): 

180 param_values = {} 

181 for param in item["params"]: 

182 param_values[param] = params[param] 

183 # Set the cookie 

184 resp.set_cookie( 

185 "param_redirect", 

186 json.dumps( 

187 {"endpoint": item["endpoint"], "params": param_values} 

188 ), 

189 # Set expiration for 10 days in the future 

190 expires=datetime.now() + timedelta(days=10), 

191 secure=True, 

192 httponly=True, 

193 ) 

194 

195 return resp 

196 

197 

198def param_redirect_exec(req, make_response, redirect): 

199 """ 

200 Function that returns a response, redirecting based on 

201 a matched cookie 

202 """ 

203 # Get cookie data 

204 encoded_redirect_data = req.cookies.get("param_redirect") 

205 

206 if encoded_redirect_data: 

207 redirect_data = json.loads(encoded_redirect_data) 

208 # Only redirect if the current path matches the redirect endpoint 

209 if req.path == redirect_data["endpoint"]: 

210 query_string = urlencode(redirect_data["params"]) 

211 response = make_response( 

212 redirect(f"{redirect_data['endpoint']}?{query_string}") 

213 ) 

214 response.set_cookie( 

215 "param_redirect", "", expires=0, secure=True, httponly=True 

216 ) 

217 return response 

218 return None 

219 

220 

221def get_csp_as_str(csp={}): 

222 csp_str = "" 

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

224 csp_value = " ".join(values) 

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

226 return csp_str.strip() 

227 

228 

229def humanize_date(date_str): 

230 if not date_str: 

231 return "" 

232 date_obj = parser.parse(date_str) 

233 return date_obj.strftime("%-d %B %Y") 

234 

235 

236def format_solution_status(status): 

237 if not status: 

238 return status 

239 return status.replace("_", " ").title() 

240 

241 

242def get_solution_form_value( 

243 form_data, solution, form_field_name, solution_path=None, default="" 

244): 

245 """ 

246 When updating a solution, get form field value 

247 When first loading edit form, shows current solution values 

248 When re-displaying form after validation errors, 

249 shows user's submitted values 

250 

251 E.g.: 

252 get_solution_form_value(form_data, solution, 'title') 

253 # checks: form_data.title -> solution.title -> "" 

254 """ 

255 if form_data and form_field_name in form_data: 

256 return form_data.get(form_field_name, default) 

257 

258 if solution: 

259 path = solution_path if solution_path else form_field_name 

260 

261 current_value = solution 

262 for key in path.split("."): 

263 if isinstance(current_value, dict) and key in current_value: 

264 current_value = current_value[key] 

265 else: 

266 return default 

267 

268 return current_value if current_value is not None else default 

269 

270 return default