Coverage for paperless/serialisers.py: 80%

1import logging 

2import os 

3from io import BytesIO 

4 

5import magic 

6from allauth.mfa.adapter import get_adapter as get_mfa_adapter 

7from allauth.mfa.models import Authenticator 

8from allauth.mfa.totp.internal.auth import TOTP 

9from allauth.socialaccount.models import SocialAccount 

10from allauth.socialaccount.models import SocialApp 

11from django.conf import settings 

12from django.contrib.auth.models import Group 

13from django.contrib.auth.models import Permission 

14from django.contrib.auth.models import User 

15from django.contrib.auth.password_validation import validate_password 

16from django.core.files.uploadedfile import InMemoryUploadedFile 

17from django.core.files.uploadedfile import UploadedFile 

18from PIL import Image 

19from rest_framework import serializers 

20from rest_framework.authtoken.serializers import AuthTokenSerializer 

21 

22from paperless.models import ApplicationConfiguration 

23from paperless.network import validate_outbound_http_url 

24from paperless.validators import reject_dangerous_svg 

25from paperless.validators import validate_raster_image 

26from paperless_mail.serialisers import ObfuscatedPasswordField 

27 

28logger = logging.getLogger("paperless.settings") 

29 

30 

31class PasswordValidationMixin: 

32 def _has_real_password(self, value: str | None) -> bool: 

33 return bool(value) and value.replace("*", "") != "" 

34 

35 def validate_password(self, value: str) -> str: 

36 if not self._has_real_password(value): 

37 return value 

38 

39 request = self.context.get("request") if hasattr(self, "context") else None 

40 user = self.instance or ( 

41 request.user if request and hasattr(request, "user") else None 

42 ) 

43 validate_password(value, user) # raise ValidationError if invalid 

44 

45 return value 

46 

47 

48class PaperlessAuthTokenSerializer(AuthTokenSerializer): 

49 code = serializers.CharField( 

50 label="MFA Code", 

51 write_only=True, 

52 required=False, 

53 ) 

54 

55 def validate(self, attrs): 

56 attrs = super().validate(attrs) 

57 user = attrs.get("user") 

58 code = attrs.get("code") 

59 mfa_adapter = get_mfa_adapter() 

60 if mfa_adapter.is_mfa_enabled(user): 

61 if not code: 

62 raise serializers.ValidationError( 

63 "MFA code is required", 

64 ) 

65 authenticator = Authenticator.objects.get( 

66 user=user, 

67 type=Authenticator.Type.TOTP, 

68 ) 

69 if not TOTP(instance=authenticator).validate_code( 

70 code, 

71 ): 

72 raise serializers.ValidationError( 

73 "Invalid MFA code", 

74 ) 

75 return attrs 

76 

77 

78class UserSerializer(PasswordValidationMixin, serializers.ModelSerializer[User]): 

79 password = ObfuscatedPasswordField(required=False) 

80 user_permissions = serializers.SlugRelatedField( 

81 many=True, 

82 queryset=Permission.objects.exclude(content_type__app_label="admin"), 

83 slug_field="codename", 

84 required=False, 

85 ) 

86 inherited_permissions = serializers.SerializerMethodField() 

87 is_mfa_enabled = serializers.SerializerMethodField() 

88 

89 def get_is_mfa_enabled(self, user: User) -> bool: 

90 mfa_adapter = get_mfa_adapter() 

91 return mfa_adapter.is_mfa_enabled(user) 

92 

93 class Meta: 

94 model = User 

95 fields = ( 

96 "id", 

97 "username", 

98 "email", 

99 "password", 

100 "first_name", 

101 "last_name", 

102 "date_joined", 

103 "is_staff", 

104 "is_active", 

105 "is_superuser", 

106 "groups", 

107 "user_permissions", 

108 "inherited_permissions", 

109 "is_mfa_enabled", 

110 ) 

111 

112 def get_inherited_permissions(self, obj) -> list[str]: 

113 return obj.get_group_permissions() 

114 

115 def update(self, instance, validated_data): 

116 password = validated_data.pop("password", None) 

117 if self._has_real_password(password): 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 instance.set_password(password) 

119 instance.save() 

120 

121 super().update(instance, validated_data) 

122 return instance 

123 

124 def create(self, validated_data): 

125 groups = None 

126 if "groups" in validated_data: 

127 groups = validated_data.pop("groups") 

128 user_permissions = None 

129 if "user_permissions" in validated_data: 

130 user_permissions = validated_data.pop("user_permissions") 

131 password = validated_data.pop("password", None) 

132 user = User.objects.create(**validated_data) 

133 # set groups 

134 if groups: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 user.groups.set(groups) 

136 # set permissions 

137 if user_permissions: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true

138 user.user_permissions.set(user_permissions) 

139 # set password 

140 if self._has_real_password(password): 

141 user.set_password(password) 

142 user.save() 

143 return user 

144 

145 

146class GroupSerializer(serializers.ModelSerializer[Group]): 

147 permissions = serializers.SlugRelatedField( 

148 many=True, 

149 queryset=Permission.objects.exclude(content_type__app_label="admin"), 

150 slug_field="codename", 

151 ) 

152 

153 class Meta: 

154 model = Group 

155 fields = ( 

156 "id", 

157 "name", 

158 "permissions", 

159 ) 

160 

161 

162class SocialAccountSerializer(serializers.ModelSerializer[SocialAccount]): 

163 name = serializers.SerializerMethodField() 

164 

165 class Meta: 

166 model = SocialAccount 

167 fields = ( 

168 "id", 

169 "provider", 

170 "name", 

171 ) 

172 

173 def get_name(self, obj: SocialAccount) -> str: 

174 try: 

175 return obj.get_provider_account().to_str() 

176 except SocialApp.DoesNotExist: 

177 return "Unknown App" 

178 

179 

180class ProfileSerializer(PasswordValidationMixin, serializers.ModelSerializer[User]): 

181 email = serializers.EmailField(allow_blank=True, required=False) 

182 password = ObfuscatedPasswordField(required=False, allow_null=False) 

183 auth_token = serializers.SlugRelatedField(read_only=True, slug_field="key") 

184 social_accounts = SocialAccountSerializer( 

185 many=True, 

186 read_only=True, 

187 source="socialaccount_set", 

188 ) 

189 is_mfa_enabled = serializers.SerializerMethodField() 

190 has_usable_password = serializers.SerializerMethodField() 

191 

192 def get_is_mfa_enabled(self, user: User) -> bool: 

193 mfa_adapter = get_mfa_adapter() 

194 return mfa_adapter.is_mfa_enabled(user) 

195 

196 def get_has_usable_password(self, user: User) -> bool: 

197 return user.has_usable_password() 

198 

199 class Meta: 

200 model = User 

201 fields = ( 

202 "email", 

203 "password", 

204 "first_name", 

205 "last_name", 

206 "auth_token", 

207 "social_accounts", 

208 "has_usable_password", 

209 "is_mfa_enabled", 

210 ) 

211 

212 

213class ApplicationConfigurationSerializer( 

214 serializers.ModelSerializer[ApplicationConfiguration], 

215): 

216 externally_configured_variables = serializers.SerializerMethodField() 

217 user_args = serializers.JSONField(binary=True, allow_null=True) 

218 barcode_tag_mapping = serializers.JSONField(binary=True, allow_null=True) 

219 llm_embedding_api_key = ObfuscatedPasswordField( 

220 required=False, 

221 allow_null=True, 

222 max_length=1024, 

223 ) 

224 llm_api_key = ObfuscatedPasswordField( 

225 required=False, 

226 allow_null=True, 

227 max_length=1024, 

228 ) 

229 remote_ocr_api_key = ObfuscatedPasswordField( 

230 required=False, 

231 allow_null=True, 

232 max_length=1024, 

233 ) 

234 

235 OBFUSCATED_FIELDS = ( 

236 "llm_embedding_api_key", 

237 "llm_api_key", 

238 "remote_ocr_api_key", 

239 ) 

240 

241 def get_externally_configured_variables( 

242 self, 

243 instance: ApplicationConfiguration, 

244 ) -> list[str]: 

245 return sorted(name for name in os.environ if name.startswith("PAPERLESS_")) 

246 

247 @staticmethod 

248 def _require_json_object(field: str, value: object) -> None: 

249 if value is not None and not isinstance(value, dict): 

250 raise serializers.ValidationError(f"{field} must be a JSON object.") 

251 

252 def validate_user_args(self, value): 

253 self._require_json_object("user_args", value) 

254 return value 

255 

256 def validate_barcode_tag_mapping(self, value): 

257 self._require_json_object("barcode_tag_mapping", value) 

258 # Each value is the regex substitute applied to a matching barcode 

259 if value is not None and not all(isinstance(v, str) for v in value.values()): 

260 raise serializers.ValidationError( 

261 "barcode_tag_mapping values must be strings.", 

262 ) 

263 return value 

264 

265 def run_validation(self, data): 

266 # Empty strings treated as None to avoid unexpected behavior 

267 if "user_args" in data and data["user_args"] == "": 

268 data["user_args"] = None 

269 if "barcode_tag_mapping" in data and data["barcode_tag_mapping"] == "": 

270 data["barcode_tag_mapping"] = None 

271 if "language" in data and data["language"] == "": 

272 data["language"] = None 

273 if "llm_output_language" in data and data["llm_output_language"] == "": 

274 data["llm_output_language"] = None 

275 for field in self.OBFUSCATED_FIELDS: 

276 if field in data and data[field] is not None: 

277 if data[field] == "": 

278 data[field] = None 

279 # Not a real value, don't overwrite the stored one 

280 elif len(data[field].replace("*", "")) == 0: 

281 del data[field] 

282 return super().run_validation(data) 

283 

284 def update(self, instance, validated_data): 

285 if instance.app_logo and "app_logo" in validated_data: 285 ↛ 286line 285 didn't jump to line 286 because the condition on line 285 was never true

286 instance.app_logo.delete() 

287 return super().update(instance, validated_data) 

288 

289 def _sanitize_raster_image(self, file: UploadedFile) -> UploadedFile: 

290 try: 

291 data = BytesIO() 

292 image = Image.open(file) 

293 image.save(data, format=image.format) 

294 data.seek(0) 

295 

296 return InMemoryUploadedFile( 

297 file=data, 

298 field_name=file.field_name, 

299 name=file.name, 

300 content_type=file.content_type, 

301 size=data.getbuffer().nbytes, 

302 charset=getattr(file, "charset", None), 

303 ) 

304 finally: 

305 image.close() 

306 

307 def validate_app_logo(self, file: UploadedFile): 

308 """ 

309 Validates and sanitizes the uploaded app logo image. Model field already restricts to 

310 jpg/png/gif/svg. 

311 """ 

312 if file: 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true

313 mime_type = magic.from_buffer(file.read(2048), mime=True) 

314 

315 if mime_type == "image/svg+xml": 

316 reject_dangerous_svg(file) 

317 else: 

318 validate_raster_image(file) 

319 

320 if mime_type in {"image/jpeg", "image/png"}: 

321 file = self._sanitize_raster_image(file) 

322 

323 return file 

324 

325 def validate_llm_endpoint(self, value: str | None) -> str | None: 

326 if not value: 

327 return value 

328 

329 try: 

330 validate_outbound_http_url( 

331 value, 

332 allow_internal=settings.LLM_ALLOW_INTERNAL_ENDPOINTS, 

333 ) 

334 except ValueError as e: 

335 raise serializers.ValidationError( 

336 f"Invalid LLM endpoint: {e.args[0]}, see logs for details", 

337 ) from e 

338 

339 return value 

340 

341 validate_llm_embedding_endpoint = validate_llm_endpoint 

342 

343 def validate_remote_ocr_endpoint(self, value: str | None) -> str | None: 

344 if not value: 

345 return value 

346 

347 try: 

348 validate_outbound_http_url( 

349 value, 

350 allow_internal=settings.REMOTE_OCR_ALLOW_INTERNAL_ENDPOINTS, 

351 ) 

352 except ValueError as e: 

353 raise serializers.ValidationError( 

354 f"Invalid remote OCR endpoint: {e.args[0]}, see logs for details", 

355 ) from e 

356 

357 return value 

358 

359 class Meta: 

360 model = ApplicationConfiguration 

361 fields = "__all__"