Coverage for postrfp / web / exception.py: 95%

1import json 

2import logging 

3import traceback 

4from pprint import pformat 

5import functools 

6 

7import webob 

8import webob.exc 

9from sqlalchemy.orm.exc import NoResultFound 

10 

11from postrfp.shared.exceptions import AuthorizationFailure 

12from postrfp.shared.constants import MIME_TYPE_VALUES 

13from postrfp.web.request import HttpRequest 

14from postrfp.templates import get_template 

15from postrfp.web.suxint import RoutingError 

16from postrfp.shared.exceptions import NotLoggedIn, UpdateConflict 

17from postrfp.model.exc import ( 

18 BusinessRuleViolation, 

19 CosmeticQuestionEditViolation, 

20 QuestionnaireStructureException, 

21 ValidationFailure, 

22 DuplicateDataProvided, 

23 DuplicateQuestionDefinition, 

24) 

25 

26log = logging.getLogger(__name__) 

27 

28 

29def sanitize_environ(environ_dict): 

30 """ 

31 Create a sanitized copy of the environ dict with sensitive values redacted. 

32 

33 This prevents passwords, API keys, tokens, and other secrets from appearing 

34 in log files while preserving useful debugging information. 

35 """ 

36 # Keys that commonly contain sensitive information 

37 SENSITIVE_KEYS = { 

38 "password", 

39 "passwd", 

40 "pwd", 

41 "secret", 

42 "token", 

43 "api_key", 

44 "apikey", 

45 "auth", 

46 "authorization", 

47 "session", 

48 "cookie", 

49 "private_key", 

50 "private", 

51 "credentials", 

52 "credential", 

53 } 

54 

55 sanitized = {} 

56 for key, value in environ_dict.items(): 

57 key_lower = key.lower() 

58 # Check if any sensitive keyword appears in the key name 

59 if any(sensitive in key_lower for sensitive in SENSITIVE_KEYS): 

60 sanitized[key] = "***REDACTED***" 

61 else: 

62 sanitized[key] = value 

63 

64 return sanitized 

65 

66 

67def render_error( 

68 request: HttpRequest, 

69 response_class, 

70 message=None, 

71 title="Server Error", 

72 errors=None, 

73) -> webob.Response: 

74 try: 

75 if request.prefers_json: 

76 # Assume a browswer ajax call - prepare response that our js client understands 

77 err_doc = dict(error=title, description=message) 

78 if errors is not None: 

79 if hasattr(errors, "as_dict"): 

80 err_doc["errors"] = errors.as_dict() 

81 elif isinstance(errors, (list, dict)): 

82 err_doc["errors"] = errors 

83 else: 

84 err_doc["errors"] = str(errors) 

85 res_js = json.dumps(err_doc) 

86 return response_class( 

87 body=res_js, charset="utf-8", content_type=str("application/json") 

88 ) 

89 

90 elif request.accept.accepts_html: 

91 # Attempt to produce a reasonable looking error message 

92 template = get_template("error.html") 

93 html_output = template.render( 

94 message=message, title=title, request=request, errors=errors 

95 ) 

96 return response_class(body=html_output) 

97 

98 elif request.accept.acceptable_offers(offers=MIME_TYPE_VALUES): 

99 return response_class(body=f"Error: {message}") 

100 

101 else: 

102 raise Exception( 

103 f"Unable to determine appropriate response type for {request.accept}" 

104 ) 

105 

106 except Exception as exc: 

107 log_exception(request, exc, logging.ERROR) 

108 m = "An unhandled error occurred. Our team is aware of the problem and is addressing it." 

109 return webob.exc.HTTPError(m) 

110 

111 

112def log_exception(request, exception, severity): 

113 """Format and log exception""" 

114 

115 stack_trace = traceback.format_exc() # Note this formats the 'current' exception 

116 environ = pformat(sanitize_environ(request.environ)) 

117 

118 log_msg = "%s: %s\n\n%s %s\n\n%s\n%s" % ( 

119 exception.__class__.__name__, 

120 str(exception), 

121 request.method, 

122 request.path_url, 

123 stack_trace, 

124 environ, 

125 ) 

126 

127 log.log(severity, log_msg) 

128 

129 

130def ex_message(ex, default="No further details"): 

131 if hasattr(ex, "message"): 

132 return ex.message 

133 if hasattr(ex, "detail"): 

134 return ex.detail 

135 return default 

136 

137 

138def resolve_exception(req: HttpRequest, e: Exception): 

139 """Map Exceptions to HTTP Responses and write Log message""" 

140 

141 # By default log exceptions with severity of INFO 

142 severity = logging.INFO 

143 error = functools.partial(render_error, req) 

144 

145 match e: 

146 case webob.exc.HTTPRedirection(): 

147 return e 

148 

149 case NotLoggedIn() | webob.exc.HTTPUnauthorized(): 

150 response = error( 

151 webob.exc.HTTPUnauthorized, message=ex_message(e), title="Not Logged In" 

152 ) 

153 

154 case NoResultFound() | RoutingError() | webob.exc.HTTPNotFound(): 

155 response = error( 

156 webob.exc.HTTPNotFound, 

157 "The resource requested is not present on the server", 

158 title="Not Found", 

159 ) 

160 

161 case AuthorizationFailure() | webob.exc.HTTPForbidden(): 

162 severity = logging.WARN 

163 msg = "The action you have attempted is forbidden: %s" % ex_message(e) 

164 response = error( 

165 webob.exc.HTTPForbidden, 

166 message=msg, 

167 title="Forbidden Action", 

168 errors=getattr(e, "errors", None), 

169 ) 

170 

171 case ValidationFailure(): 

172 msg = "The request could not be processed due to invalid data provided." 

173 response = error( 

174 webob.exc.HTTPBadRequest, 

175 message=msg, 

176 title="Invalid Information Submitted", 

177 errors=e.errors_list, 

178 ) 

179 

180 case ValueError(): 

181 msg = "The request could not be processed due to invalid data provided." 

182 response = error( 

183 webob.exc.HTTPBadRequest, 

184 message=msg, 

185 title="Invalid Information Submitted", 

186 errors=[str(e)], 

187 ) 

188 

189 case UpdateConflict(): 

190 msg = ex_message( 

191 e, default="The resource has been modified since you last retrieved it." 

192 ) 

193 response = error( 

194 webob.exc.HTTPPreconditionFailed, 

195 message=msg, 

196 title="Precondition Failed - Update Conflict", 

197 ) 

198 

199 case DuplicateDataProvided(): 

200 msg = ex_message(e, default="Duplicate Data provided") 

201 response = error(webob.exc.HTTPConflict, message=msg, title="Data Conflict") 

202 

203 case CosmeticQuestionEditViolation(): 

204 msg = ex_message( 

205 e, default="Cannot delete question elements that have related answers" 

206 ) 

207 response = error( 

208 webob.exc.HTTPConflict, message=msg, title="Data Integrity Violation" 

209 ) 

210 

211 case webob.exc.HTTPBadRequest(): 

212 msg = ex_message( 

213 e, default="Bad Request - probably an outdated or corrupted link" 

214 ) 

215 response = error( 

216 webob.exc.HTTPBadRequest, 

217 message=msg, 

218 title="Bad Request - possibly an outdate or corrupt link", 

219 ) 

220 

221 case QuestionnaireStructureException(): 

222 msg = ex_message(e, default="Invalid questionnaire structure") 

223 response = error( 

224 webob.exc.HTTPBadRequest, 

225 message=msg, 

226 title="Questionnaire Structure Violation", 

227 ) 

228 

229 case BusinessRuleViolation(): 

230 msg = ex_message(e, default="Illegal Action: %s" % e.message) 

231 response = error( 

232 webob.exc.HTTPBadRequest, message=msg, title="Illegal Action" 

233 ) 

234 

235 case DuplicateQuestionDefinition(): 

236 msg = ex_message( 

237 e, default="Question has already been shared to this project" 

238 ) 

239 response = error( 

240 webob.exc.HTTPBadRequest, 

241 message=msg, 

242 title="Bad Request - Section Data Integrity Violation", 

243 ) 

244 

245 case webob.exc.HTTPError(): 

246 response = error(type(e), e) 

247 

248 case _: 

249 severity = logging.ERROR 

250 msg = "An unexpected error occurred processing your request." 

251 msg += " Our engineers have been informed so there is no need for action on your part." 

252 msg += " It may help to reload the page." 

253 response = error(webob.exc.HTTPServerError, msg) 

254 

255 log_exception(req, e, severity) 

256 

257 return response