|
5 | 5 | import unittest |
6 | 6 | from typing import TYPE_CHECKING, Any, ClassVar, cast |
7 | 7 |
|
8 | | -from maxminddb import MODE_MEMORY, open_database |
| 8 | +from maxminddb import MODE_MEMORY, MODE_MMAP_EXT, open_database |
9 | 9 | from maxminddb.decoder import Decoder |
10 | 10 | from maxminddb.errors import InvalidDatabaseError |
11 | 11 |
|
| 12 | +try: |
| 13 | + import maxminddb.extension as _extension |
| 14 | +except ImportError: |
| 15 | + _extension = None # type: ignore[assignment] |
| 16 | + |
12 | 17 | if TYPE_CHECKING: |
13 | 18 | from _typeshed import SizedBuffer |
14 | 19 |
|
@@ -449,3 +454,72 @@ def test_normal_record_still_decodes(self) -> None: |
449 | 454 | ) |
450 | 455 | self.assertEqual(record["utf8_string"], "unicode! ☯ - ♫") |
451 | 456 | self.assertEqual(record["bytes"], b"\x00\x00\x00*") |
| 457 | + |
| 458 | + |
| 459 | +def _has_extension() -> bool: |
| 460 | + return _extension is not None and hasattr(_extension, "Reader") |
| 461 | + |
| 462 | + |
| 463 | +# The patched libmaxminddb reports its decoder resource limits through this |
| 464 | +# text (MMDB_DECODER_LIMIT_ERROR). A libmaxminddb without the fix decodes the |
| 465 | +# DoS fixtures instead, so the tests below skip rather than run the extension's |
| 466 | +# decoder out of memory. |
| 467 | +_EXTENSION_LIMIT_MESSAGE = "exceeds the configured resource limits" |
| 468 | + |
| 469 | + |
| 470 | +@unittest.skipUnless(_has_extension(), "C extension not available") |
| 471 | +class TestExtensionResourceLimits(unittest.TestCase): |
| 472 | + """DoS-fixture checks for the C extension's libmaxminddb decoder. |
| 473 | +
|
| 474 | + The extension decodes through libmaxminddb, so these limits live in that |
| 475 | + library, not in the pure-Python decoder that TestDecoderResourceLimits |
| 476 | + covers. The checks run only when the linked libmaxminddb enforces the |
| 477 | + limits and skip otherwise; see setUp. |
| 478 | + """ |
| 479 | + |
| 480 | + @staticmethod |
| 481 | + def _lookup(filename: str, ip: str = "0.0.0.1") -> object: |
| 482 | + # MODE_MMAP_EXT forces the C extension. Each DoS fixture resolves any |
| 483 | + # IPv4 address to its single crafted record. |
| 484 | + with open_database( |
| 485 | + f"{_TEST_DATA_DIR}/{filename}", |
| 486 | + mode=MODE_MMAP_EXT, |
| 487 | + ) as reader: |
| 488 | + return reader.get(ip) |
| 489 | + |
| 490 | + def setUp(self) -> None: |
| 491 | + # Probe with a fixture one byte over the 2 MiB payload limit. A patched |
| 492 | + # libmaxminddb rejects it with the decoder-limit message. An older one |
| 493 | + # decodes it, which is only about 2 MiB and so safe, but means the large |
| 494 | + # DoS fixtures below would exhaust memory, so skip instead of running |
| 495 | + # them. |
| 496 | + try: |
| 497 | + self._lookup("MaxMind-DB-test-decoder-payload-limit-over.mmdb") |
| 498 | + except InvalidDatabaseError as exc: |
| 499 | + if _EXTENSION_LIMIT_MESSAGE in str(exc): |
| 500 | + return |
| 501 | + self.skipTest( |
| 502 | + "linked libmaxminddb predates the decoder resource limits " |
| 503 | + "(needs the fix that adds MMDB_DECODER_LIMIT_ERROR)", |
| 504 | + ) |
| 505 | + |
| 506 | + def test_pointer_fan_out_fixture_is_rejected(self) -> None: |
| 507 | + # A full database whose record nests arrays of pointers to the level |
| 508 | + # below, the classic 2**depth fan-out. |
| 509 | + with self.assertRaisesRegex(InvalidDatabaseError, _EXTENSION_LIMIT_MESSAGE): |
| 510 | + self._lookup("MaxMind-DB-test-pointer-decoder-dos.mmdb") |
| 511 | + |
| 512 | + def test_payload_amplification_is_rejected(self) -> None: |
| 513 | + # An array of 8,192 pointers to one 65,535-byte value. |
| 514 | + with self.assertRaisesRegex(InvalidDatabaseError, _EXTENSION_LIMIT_MESSAGE): |
| 515 | + self._lookup("MaxMind-DB-test-payload-amplification-dos.mmdb") |
| 516 | + |
| 517 | + def test_payload_amplification_string_is_rejected(self) -> None: |
| 518 | + # The UTF-8 string variant, so the string decode path is exercised. |
| 519 | + with self.assertRaisesRegex(InvalidDatabaseError, _EXTENSION_LIMIT_MESSAGE): |
| 520 | + self._lookup("MaxMind-DB-test-payload-amplification-dos-string.mmdb") |
| 521 | + |
| 522 | + def test_payload_amplification_worst_case_is_rejected(self) -> None: |
| 523 | + # 65,535 pointers to one 65,535-byte value, exactly the value limit. |
| 524 | + with self.assertRaisesRegex(InvalidDatabaseError, _EXTENSION_LIMIT_MESSAGE): |
| 525 | + self._lookup("MaxMind-DB-test-payload-amplification-dos-worst-case.mmdb") |
0 commit comments