|
52 | 52 | class BackupsTest(ViewTestCase): |
53 | 53 | CREATE_GLOSSARIES: bool = True |
54 | 54 |
|
| 55 | + def write_tampered_component_backup( |
| 56 | + self, *, repo: str | None = None, push: str | None = None |
| 57 | + ) -> str: |
| 58 | + backup = ProjectBackup() |
| 59 | + backup.backup_project(self.project) |
| 60 | + |
| 61 | + with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_handle: |
| 62 | + temp_name = temp_handle.name |
| 63 | + |
| 64 | + with ( |
| 65 | + ZipFile(backup.filename, "r") as source_zip, |
| 66 | + ZipFile(temp_name, "w") as target_zip, |
| 67 | + ): |
| 68 | + for item in source_zip.infolist(): |
| 69 | + data = source_zip.read(item.filename) |
| 70 | + if item.filename.endswith( |
| 71 | + f"{self.component.slug}.json" |
| 72 | + ) and item.filename.startswith("components/"): |
| 73 | + component_data = json.loads(data.decode("utf-8")) |
| 74 | + if repo is not None: |
| 75 | + component_data["component"]["repo"] = repo |
| 76 | + if push is not None: |
| 77 | + component_data["component"]["push"] = push |
| 78 | + data = json.dumps(component_data).encode("utf-8") |
| 79 | + target_zip.writestr(item, data) |
| 80 | + |
| 81 | + return temp_name |
| 82 | + |
55 | 83 | def test_backup_creates_history_entry(self) -> None: |
56 | 84 | backup = ProjectBackup() |
57 | 85 |
|
@@ -355,6 +383,60 @@ def test_restore_synthesizes_source_translation_check_flags(self) -> None: |
355 | 383 |
|
356 | 384 | self.assertEqual(restored_source.check_flags, "read-only") |
357 | 385 |
|
| 386 | + def test_restore_rejects_invalid_repo_url(self) -> None: |
| 387 | + temp_name = self.write_tampered_component_backup( |
| 388 | + repo="https://private.example/repo.git" |
| 389 | + ) |
| 390 | + |
| 391 | + try: |
| 392 | + restore = ProjectBackup(temp_name) |
| 393 | + with ( |
| 394 | + patch( |
| 395 | + "weblate.utils.outbound.socket.getaddrinfo", |
| 396 | + return_value=[(0, 0, 0, "", ("127.0.0.1", 443))], |
| 397 | + ), |
| 398 | + self.assertRaises(ValidationError) as error, |
| 399 | + ): |
| 400 | + restore.validate() |
| 401 | + |
| 402 | + self.assertEqual( |
| 403 | + error.exception.message_dict, |
| 404 | + { |
| 405 | + "repo": [ |
| 406 | + "This URL is prohibited because it points to an internal or non-public address." |
| 407 | + ] |
| 408 | + }, |
| 409 | + ) |
| 410 | + finally: |
| 411 | + os.unlink(temp_name) |
| 412 | + |
| 413 | + def test_restore_rejects_invalid_push_url(self) -> None: |
| 414 | + temp_name = self.write_tampered_component_backup( |
| 415 | + push="https://private.example/push.git" |
| 416 | + ) |
| 417 | + |
| 418 | + try: |
| 419 | + restore = ProjectBackup(temp_name) |
| 420 | + with ( |
| 421 | + patch( |
| 422 | + "weblate.utils.outbound.socket.getaddrinfo", |
| 423 | + return_value=[(0, 0, 0, "", ("127.0.0.1", 443))], |
| 424 | + ), |
| 425 | + self.assertRaises(ValidationError) as error, |
| 426 | + ): |
| 427 | + restore.validate() |
| 428 | + |
| 429 | + self.assertEqual( |
| 430 | + error.exception.message_dict, |
| 431 | + { |
| 432 | + "push": [ |
| 433 | + "This URL is prohibited because it points to an internal or non-public address." |
| 434 | + ] |
| 435 | + }, |
| 436 | + ) |
| 437 | + finally: |
| 438 | + os.unlink(temp_name) |
| 439 | + |
358 | 440 | def test_create_duplicate(self) -> None: |
359 | 441 | def extract_names(qs) -> list[str]: |
360 | 442 | return list(qs.order_by("name").values_list("name", flat=True)) |
|
0 commit comments