|
2 | 2 |
|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
| 5 | +import os |
5 | 6 | from unittest.mock import AsyncMock, MagicMock, patch |
6 | 7 |
|
7 | 8 | import pytest |
@@ -594,6 +595,236 @@ def _mock_verify(app_id: str, public_pem: bytes) -> bool: |
594 | 595 | # Restore the original so teardown is clean. |
595 | 596 | reg.verify_manifest_signature = original_verify # type: ignore[method-assign] |
596 | 597 |
|
| 598 | + # ── TOCTOU refusal-path tests ────────────────────────────────────── |
| 599 | + # Each test exercises one failure mode of the install-time TOCTOU |
| 600 | + # re-verification guard. The first gate is monkeypatched to return |
| 601 | + # success so the TOCTOU guard is what actually catches the failure; |
| 602 | + # each test asserts 403 to prove the gate goes red where it counts. |
| 603 | + |
| 604 | + @pytest.mark.asyncio |
| 605 | + async def test_toctou_manifest_missing_returns_403(self, client, tmp_path): |
| 606 | + """TOCTOU re-verify returns 403 when manifest.yaml is deleted |
| 607 | + between the initial gate check and the install.""" |
| 608 | + from tinyagentos.registry import AppRegistry |
| 609 | + from tinyagentos.store_signing import generate_signing_keypair |
| 610 | + |
| 611 | + catalog_dir = tmp_path / "catalog" |
| 612 | + svc_dir = catalog_dir / "services" / "test-svc" |
| 613 | + svc_dir.mkdir(parents=True) |
| 614 | + manifest_path = svc_dir / "manifest.yaml" |
| 615 | + manifest_path.write_text( |
| 616 | + "id: test-svc\nname: Test Service\ntype: service\n" |
| 617 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 618 | + ) |
| 619 | + |
| 620 | + priv, pub = generate_signing_keypair() |
| 621 | + installed_path = tmp_path / "installed.json" |
| 622 | + installed_path.write_text("[]") |
| 623 | + reg = AppRegistry( |
| 624 | + catalog_dir=catalog_dir, |
| 625 | + installed_path=installed_path, |
| 626 | + signing_key=priv, |
| 627 | + ) |
| 628 | + reg._ensure_loaded() |
| 629 | + |
| 630 | + client._transport.app.state.registry = reg |
| 631 | + client._transport.app.state.store_signing_pubkey = pub |
| 632 | + client._transport.app.state.installed_apps = _make_installed_apps() |
| 633 | + |
| 634 | + # Make the first gate pass so the TOCTOU guard runs. |
| 635 | + reg.verify_manifest_signature = lambda aid, pem: True # type: ignore[method-assign] |
| 636 | + |
| 637 | + # Sabotage: delete the manifest from disk. |
| 638 | + os.remove(manifest_path) |
| 639 | + |
| 640 | + resp = await client.post("/api/store/install-v2", json={ |
| 641 | + "manifest_id": "test-svc", |
| 642 | + }) |
| 643 | + assert resp.status_code == 403 |
| 644 | + assert resp.json()["error"] == ( |
| 645 | + "manifest modified between signature verification and install" |
| 646 | + ) |
| 647 | + |
| 648 | + @pytest.mark.asyncio |
| 649 | + async def test_toctou_manifest_unreadable_returns_403(self, client, tmp_path): |
| 650 | + """TOCTOU re-verify returns 403 when manifest.yaml cannot be read |
| 651 | + (permissions revoked) between the initial gate and the install.""" |
| 652 | + from tinyagentos.registry import AppRegistry |
| 653 | + from tinyagentos.store_signing import generate_signing_keypair |
| 654 | + |
| 655 | + catalog_dir = tmp_path / "catalog" |
| 656 | + svc_dir = catalog_dir / "services" / "test-svc" |
| 657 | + svc_dir.mkdir(parents=True) |
| 658 | + manifest_path = svc_dir / "manifest.yaml" |
| 659 | + manifest_path.write_text( |
| 660 | + "id: test-svc\nname: Test Service\ntype: service\n" |
| 661 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 662 | + ) |
| 663 | + |
| 664 | + priv, pub = generate_signing_keypair() |
| 665 | + installed_path = tmp_path / "installed.json" |
| 666 | + installed_path.write_text("[]") |
| 667 | + reg = AppRegistry( |
| 668 | + catalog_dir=catalog_dir, |
| 669 | + installed_path=installed_path, |
| 670 | + signing_key=priv, |
| 671 | + ) |
| 672 | + reg._ensure_loaded() |
| 673 | + |
| 674 | + client._transport.app.state.registry = reg |
| 675 | + client._transport.app.state.store_signing_pubkey = pub |
| 676 | + client._transport.app.state.installed_apps = _make_installed_apps() |
| 677 | + |
| 678 | + reg.verify_manifest_signature = lambda aid, pem: True # type: ignore[method-assign] |
| 679 | + |
| 680 | + # Sabotage: revoke read permission. |
| 681 | + try: |
| 682 | + os.chmod(manifest_path, 0o000) |
| 683 | + resp = await client.post("/api/store/install-v2", json={ |
| 684 | + "manifest_id": "test-svc", |
| 685 | + }) |
| 686 | + assert resp.status_code == 403 |
| 687 | + assert resp.json()["error"] == ( |
| 688 | + "manifest modified between signature verification and install" |
| 689 | + ) |
| 690 | + finally: |
| 691 | + os.chmod(manifest_path, 0o644) # restore so tmp_path can clean up |
| 692 | + |
| 693 | + @pytest.mark.asyncio |
| 694 | + async def test_toctou_safe_load_empty_returns_403(self, client, tmp_path): |
| 695 | + """TOCTOU re-verify returns 403 when the on-disk YAML parses to |
| 696 | + an empty/falsy value (truncated or corrupted manifest).""" |
| 697 | + from tinyagentos.registry import AppRegistry |
| 698 | + from tinyagentos.store_signing import generate_signing_keypair |
| 699 | + |
| 700 | + catalog_dir = tmp_path / "catalog" |
| 701 | + svc_dir = catalog_dir / "services" / "test-svc" |
| 702 | + svc_dir.mkdir(parents=True) |
| 703 | + manifest_path = svc_dir / "manifest.yaml" |
| 704 | + manifest_path.write_text( |
| 705 | + "id: test-svc\nname: Test Service\ntype: service\n" |
| 706 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 707 | + ) |
| 708 | + |
| 709 | + priv, pub = generate_signing_keypair() |
| 710 | + installed_path = tmp_path / "installed.json" |
| 711 | + installed_path.write_text("[]") |
| 712 | + reg = AppRegistry( |
| 713 | + catalog_dir=catalog_dir, |
| 714 | + installed_path=installed_path, |
| 715 | + signing_key=priv, |
| 716 | + ) |
| 717 | + reg._ensure_loaded() |
| 718 | + |
| 719 | + client._transport.app.state.registry = reg |
| 720 | + client._transport.app.state.store_signing_pubkey = pub |
| 721 | + client._transport.app.state.installed_apps = _make_installed_apps() |
| 722 | + |
| 723 | + reg.verify_manifest_signature = lambda aid, pem: True # type: ignore[method-assign] |
| 724 | + |
| 725 | + # Sabotage: truncate the manifest to empty. |
| 726 | + manifest_path.write_text("") |
| 727 | + |
| 728 | + resp = await client.post("/api/store/install-v2", json={ |
| 729 | + "manifest_id": "test-svc", |
| 730 | + }) |
| 731 | + assert resp.status_code == 403 |
| 732 | + assert resp.json()["error"] == ( |
| 733 | + "manifest modified between signature verification and install" |
| 734 | + ) |
| 735 | + |
| 736 | + @pytest.mark.asyncio |
| 737 | + async def test_toctou_stored_sig_none_returns_403(self, client, tmp_path): |
| 738 | + """TOCTOU re-verify returns 403 when the registry has no stored |
| 739 | + signature for the manifest (signature was lost or never persisted). |
| 740 | +
|
| 741 | + The first gate sees ``stored_sig is None`` and, because the |
| 742 | + manifest is not a signing failure, allows the install through. |
| 743 | + The TOCTOU guard then re-checks on its own and blocks — proving |
| 744 | + the two gates are independently fail-closed for this case too.""" |
| 745 | + from tinyagentos.registry import AppRegistry |
| 746 | + from tinyagentos.store_signing import generate_signing_keypair |
| 747 | + |
| 748 | + catalog_dir = tmp_path / "catalog" |
| 749 | + svc_dir = catalog_dir / "services" / "test-svc" |
| 750 | + svc_dir.mkdir(parents=True) |
| 751 | + manifest_path = svc_dir / "manifest.yaml" |
| 752 | + manifest_path.write_text( |
| 753 | + "id: test-svc\nname: Test Service\ntype: service\n" |
| 754 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 755 | + ) |
| 756 | + |
| 757 | + priv, pub = generate_signing_keypair() |
| 758 | + installed_path = tmp_path / "installed.json" |
| 759 | + installed_path.write_text("[]") |
| 760 | + reg = AppRegistry( |
| 761 | + catalog_dir=catalog_dir, |
| 762 | + installed_path=installed_path, |
| 763 | + signing_key=priv, |
| 764 | + ) |
| 765 | + reg._ensure_loaded() |
| 766 | + # Clear the stored signature so both gates see None. |
| 767 | + reg._signatures.pop("test-svc", None) |
| 768 | + |
| 769 | + client._transport.app.state.registry = reg |
| 770 | + client._transport.app.state.store_signing_pubkey = pub |
| 771 | + client._transport.app.state.installed_apps = _make_installed_apps() |
| 772 | + |
| 773 | + resp = await client.post("/api/store/install-v2", json={ |
| 774 | + "manifest_id": "test-svc", |
| 775 | + }) |
| 776 | + assert resp.status_code == 403 |
| 777 | + assert resp.json()["error"] == ( |
| 778 | + "manifest modified between signature verification and install" |
| 779 | + ) |
| 780 | + |
| 781 | + @pytest.mark.asyncio |
| 782 | + async def test_toctou_signature_mismatch_returns_403(self, client, tmp_path): |
| 783 | + """TOCTOU re-verify returns 403 when the on-disk manifest has been |
| 784 | + tampered with between the initial gate check and the install.""" |
| 785 | + from tinyagentos.registry import AppRegistry |
| 786 | + from tinyagentos.store_signing import generate_signing_keypair |
| 787 | + |
| 788 | + catalog_dir = tmp_path / "catalog" |
| 789 | + svc_dir = catalog_dir / "services" / "test-svc" |
| 790 | + svc_dir.mkdir(parents=True) |
| 791 | + manifest_path = svc_dir / "manifest.yaml" |
| 792 | + manifest_path.write_text( |
| 793 | + "id: test-svc\nname: Test Service\ntype: service\n" |
| 794 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 795 | + ) |
| 796 | + |
| 797 | + priv, pub = generate_signing_keypair() |
| 798 | + installed_path = tmp_path / "installed.json" |
| 799 | + installed_path.write_text("[]") |
| 800 | + reg = AppRegistry( |
| 801 | + catalog_dir=catalog_dir, |
| 802 | + installed_path=installed_path, |
| 803 | + signing_key=priv, |
| 804 | + ) |
| 805 | + reg._ensure_loaded() |
| 806 | + |
| 807 | + client._transport.app.state.registry = reg |
| 808 | + client._transport.app.state.store_signing_pubkey = pub |
| 809 | + client._transport.app.state.installed_apps = _make_installed_apps() |
| 810 | + |
| 811 | + # Make the first gate pass so the TOCTOU guard runs. |
| 812 | + reg.verify_manifest_signature = lambda aid, pem: True # type: ignore[method-assign] |
| 813 | + |
| 814 | + # Sabotage: tamper with the manifest on disk. |
| 815 | + manifest_path.write_text( |
| 816 | + "id: test-svc\nname: EVIL Service\ntype: service\n" |
| 817 | + "version: \"1.0\"\ninstall:\n method: download\n", |
| 818 | + ) |
| 819 | + |
| 820 | + resp = await client.post("/api/store/install-v2", json={ |
| 821 | + "manifest_id": "test-svc", |
| 822 | + }) |
| 823 | + assert resp.status_code == 403 |
| 824 | + assert resp.json()["error"] == ( |
| 825 | + "manifest modified between signature verification and install" |
| 826 | + ) |
| 827 | + |
597 | 828 | @pytest.mark.asyncio |
598 | 829 | async def test_no_signing_key_skips_verification(self, client): |
599 | 830 | """When store_signing_pubkey is not set, the signing gate is |
|
0 commit comments