Coverage for an_website / utils / decorators.py: 80.000%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""A module with useful decorators.""" 

15 

16 

17import contextlib 

18import logging 

19from base64 import b64decode 

20from collections.abc import Callable, Mapping 

21from functools import wraps 

22from typing import Any, Final, ParamSpec, TypeVar, cast, overload 

23 

24from tornado.web import RequestHandler 

25 

26from .token import InvalidTokenError, parse_token 

27from .utils import Permission, anonymize_ip 

28 

29Default = TypeVar("Default") 

30Args = ParamSpec("Args") 

31Ret = TypeVar("Ret") 

32 

33 

34def keydecode( 

35 token: str, 

36 api_secrets: Mapping[str | None, Permission], 

37 token_secret: str | bytes | None, 

38) -> None | Permission: 

39 """Decode a key.""" 

40 tokens: list[str] = [token] 

41 decoded: str | None 

42 try: 

43 decoded = b64decode(token).decode("UTF-8") 

44 except ValueError: 

45 decoded = None 

46 else: 

47 tokens.append(decoded) 

48 if token_secret: 

49 for _ in tokens: 

50 with contextlib.suppress(InvalidTokenError): 

51 return parse_token(_, secret=token_secret).permissions 

52 if decoded is None: 

53 return None 

54 return api_secrets.get(decoded) 

55 

56 

57def is_authorized( 

58 inst: RequestHandler, 

59 permission: Permission, 

60 allow_cookie_auth: bool = True, 

61) -> None | bool: 

62 """Check whether the request is authorized.""" 

63 keys: dict[str | None, Permission] = inst.settings.get( 

64 "TRUSTED_API_SECRETS", {} 

65 ) 

66 token_secret: str | bytes | None = inst.settings.get("AUTH_TOKEN_SECRET") 

67 

68 permissions: tuple[None | Permission, ...] = ( # TODO: CLEAN-UP THIS MESS!! 

69 *( 

70 ( 

71 keydecode(_[7:], keys, token_secret) 

72 if _.lower().startswith("bearer ") 

73 else keys.get(_) 

74 ) 

75 for _ in inst.request.headers.get_list("Authorization") 

76 ), 

77 *( 

78 keydecode(_, keys, token_secret) 

79 for _ in inst.get_arguments("access_token") 

80 ), 

81 *(keys.get(_) for _ in inst.get_arguments("key")), 

82 ( 

83 keydecode( 

84 inst.get_cookie("access_token", ""), 

85 keys, 

86 token_secret, 

87 ) 

88 if allow_cookie_auth 

89 else None 

90 ), 

91 keys.get(inst.get_cookie("key", None)) if allow_cookie_auth else None, 

92 ) 

93 

94 if all(perm is None for perm in permissions): 

95 return None 

96 

97 result = Permission(0) 

98 for perm in permissions: 

99 if perm: 

100 result |= perm 

101 

102 return permission in result 

103 

104 

105_DEFAULT_VALUE: Final = object() 

106 

107 

108@overload 

109def requires( 

110 *perms: Permission, 

111 return_instead_of_finishing: Default, 

112 allow_cookie_auth: bool = True, 

113) -> Callable[[Callable[Args, Ret]], Callable[Args, Ret | Default]]: ... 

114 

115 

116@overload 

117def requires( 

118 *perms: Permission, 

119 allow_cookie_auth: bool = True, 

120) -> Callable[[Callable[Args, Ret]], Callable[Args, Ret]]: ... 

121 

122 

123def requires( 

124 *perms: Permission, 

125 return_instead_of_finishing: Any = _DEFAULT_VALUE, 

126 allow_cookie_auth: bool = True, 

127) -> Callable[[Callable[Args, Ret]], Callable[Args, Any]]: 

128 """Handle required permissions.""" 

129 permissions = Permission(0) 

130 for perm in perms: 

131 permissions |= perm 

132 

133 finish_with_error = return_instead_of_finishing is _DEFAULT_VALUE 

134 

135 def internal(method: Callable[Args, Ret]) -> Callable[Args, Any]: 

136 method.required_perms = permissions # type: ignore[attr-defined] 

137 logger = logging.getLogger(f"{method.__module__}.{method.__qualname__}") 

138 

139 wraps(method) 

140 

141 @wraps(method) 

142 def wrapper(*args: Args.args, **kwargs: Args.kwargs) -> Any: 

143 instance = args[0] 

144 if not isinstance(instance, RequestHandler): 

145 raise TypeError(f"Instance has invalid type {type(instance)}") 

146 authorized = is_authorized(instance, permissions, allow_cookie_auth) 

147 if not authorized: 

148 if not finish_with_error: 

149 return cast(Ret, return_instead_of_finishing) 

150 logger.warning( 

151 "Unauthorized access to %s from %s", 

152 instance.request.path, 

153 anonymize_ip(instance.request.remote_ip), 

154 ) 

155 instance.clear() 

156 instance.set_header("WWW-Authenticate", "Bearer") 

157 status = 401 if authorized is None else 403 

158 instance.set_status(status) 

159 instance.write_error(status, **kwargs) 

160 return None 

161 

162 return method(*args, **kwargs) 

163 

164 return wrapper 

165 

166 return internal 

167 

168 

169@overload 

170def requires_settings( 

171 *settings: str, 

172 return_: Default, 

173) -> Callable[[Callable[Args, Ret]], Callable[Args, Ret | Default]]: ... 

174 

175 

176@overload 

177def requires_settings( 

178 *settings: str, 

179 status_code: int, 

180) -> Callable[[Callable[Args, Ret]], Callable[Args, Ret | None]]: ... 

181 

182 

183def requires_settings( 

184 *settings: str, 

185 return_: Any = _DEFAULT_VALUE, 

186 status_code: int | None = None, 

187) -> Callable[[Callable[Args, Ret]], Callable[Args, Any]]: 

188 """Require some settings to execute a method.""" 

189 finish_with_error = return_ is _DEFAULT_VALUE 

190 if not finish_with_error and isinstance(status_code, int): 

191 raise ValueError("return_ and finish_status specified") 

192 if finish_with_error and status_code is None: 

193 status_code = 503 

194 

195 def internal(method: Callable[Args, Ret]) -> Callable[Args, Any]: 

196 logger = logging.getLogger(f"{method.__module__}.{method.__qualname__}") 

197 

198 @wraps(method) 

199 def wrapper(*args: Args.args, **kwargs: Args.kwargs) -> Any: 

200 instance = args[0] 

201 if not isinstance(instance, RequestHandler): 

202 raise TypeError(f"Instance has invalid type {type(instance)}") 

203 missing = [ 

204 setting 

205 for setting in settings 

206 if instance.settings.get(setting) is None 

207 ] 

208 if missing: 

209 if not finish_with_error: 

210 return cast(Ret, return_) 

211 logger.warning( 

212 "Missing settings %s for request to %s", 

213 ", ".join(missing), 

214 instance.request.path, 

215 ) 

216 instance.send_error(cast(int, status_code)) 

217 return None 

218 for setting in settings: 

219 kwargs[setting.lower()] = instance.settings[setting] 

220 return method(*args, **kwargs) 

221 

222 return wrapper 

223 

224 return internal 

225 

226 

227def get_setting_or_default( 

228 setting: str, 

229 default: Any, 

230) -> Callable[[Callable[Args, Ret]], Callable[Args, Ret]]: 

231 """Require some settings to execute a method.""" 

232 

233 def internal(method: Callable[Args, Ret]) -> Callable[Args, Ret]: 

234 @wraps(method) 

235 def wrapper(*args: Args.args, **kwargs: Args.kwargs) -> Ret: 

236 instance = args[0] 

237 if not isinstance(instance, RequestHandler): 

238 raise TypeError(f"Instance has invalid type {type(instance)}") 

239 kwargs[setting.lower()] = instance.settings.get(setting, default) 

240 return method(*args, **kwargs) 

241 

242 return wrapper 

243 

244 return internal