diff --git a/SOURCE_MANIFEST.json b/SOURCE_MANIFEST.json index 065eddc..ee37916 100644 --- a/SOURCE_MANIFEST.json +++ b/SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schema_version": "thoth-source-preview-manifest-v1", - "created_at": "2026-10-18T07:15:13.948726+00:00", + "created_at": "2026-10-19T10:25:51.945157+00:00", "license": "MIT", "git_history_included": false, "full_suite_verified": false, @@ -1948,13 +1948,13 @@ }, { "path": "docs/KNOWN_LIMITATIONS.ko.md", - "sha256": "84e4fe631a04704d90a354346240b016a273eedb5de9b78fe819e2ddb58be726", - "bytes": 7334 + "sha256": "d8b0bd6598ae913e90c5144b163d6938009845e06c179ca21e56f0cd24b4e33f", + "bytes": 8165 }, { "path": "docs/KNOWN_LIMITATIONS.md", - "sha256": "d7ff5b62fc5d3f0551165924b13889709d84b2778d8bf1cd2bcd9885cdcf807c", - "bytes": 6446 + "sha256": "1d782ae99f440fdd981ae120ab33592b8f90d6146d4b47f07d2bb314fac3398e", + "bytes": 7100 }, { "path": "docs/KNOWN_TEST_FAILURES.json", @@ -2508,8 +2508,8 @@ }, { "path": "migrations/env.py", - "sha256": "65cdf486fe3014184e90e941cba3a561536db7b5a035446729a854fc0c2dc869", - "bytes": 1788 + "sha256": "feb20ec5fb10d3cb726e07d14b1b6c5ae413a18b1072f5de787c930ad3555c54", + "bytes": 1820 }, { "path": "migrations/script.py.mako", @@ -3278,8 +3278,13 @@ }, { "path": "src/thoth/adapters/connectors/anonymous_browser.py", - "sha256": "ccf46824afa85299324d0399888209e947be355f988281bfedeeb2fb2d9321b2", - "bytes": 9421 + "sha256": "b897c21de59bd28a96af051689ae7afa2ee5da9e1d5914278048ab4cfe85b43a", + "bytes": 8152 + }, + { + "path": "src/thoth/adapters/connectors/anonymous_browser_lifecycle.py", + "sha256": "03cf2f12efeb21f5158c25352042571abe471cd45f83a95bd014808f9cc9d071", + "bytes": 3579 }, { "path": "src/thoth/adapters/connectors/browser_document_capture.py", @@ -3423,8 +3428,8 @@ }, { "path": "src/thoth/adapters/http/app.py", - "sha256": "0e73d5cf7c516af7d2e68c1bda3f27da818c7a5bb928f08425bb1653b5d12ca0", - "bytes": 20695 + "sha256": "a56006788fef587dc53e012d27bdf41fc09c67e6fe93f5198ea758abf7f4e1e9", + "bytes": 20812 }, { "path": "src/thoth/adapters/http/async_routes.py", @@ -3446,6 +3451,11 @@ "sha256": "4b88e1d45bc8927942dccb85bcceece95e8a570d32b8cef3814585cfafc29ad9", "bytes": 5808 }, + { + "path": "src/thoth/adapters/http/local_request_guard.py", + "sha256": "c9c961d282adb9dcbd0f5fea6373224c45ab25f9407d2afdbdd320bd2fdac8fa", + "bytes": 4455 + }, { "path": "src/thoth/adapters/improvement.py", "sha256": "1dfa623054248b40b07b0e29445b41df9854a674cb5cc716ef235f4725bb0f85", @@ -3903,8 +3913,8 @@ }, { "path": "src/thoth/adapters/storage/action.py", - "sha256": "4fdc09b09b91281a4f1a06145bb163a6fb10a68bfbb12abed0e8fed8a87e66ab", - "bytes": 15732 + "sha256": "26f9db308116e146d36c5cb173c6e49bdce714e37f77a98ecfe2c41aa35ad1f9", + "bytes": 16247 }, { "path": "src/thoth/adapters/storage/action_schema.py", @@ -4318,8 +4328,8 @@ }, { "path": "src/thoth/application/commands/executions_full.py", - "sha256": "6c841c7dcab63751917496acb0c6bec1cf4e565f7f2ebefa16037ab9f1de5fbe", - "bytes": 24798 + "sha256": "aedecedac848b08aeb612a32e08df172396440571493ae6063311e2f275f561b", + "bytes": 25722 }, { "path": "src/thoth/application/commands/executions_inputs.py", @@ -4758,13 +4768,13 @@ }, { "path": "src/thoth/application/services/authorization_consumption.py", - "sha256": "7bfa561baa7b4d86aaac46b4243c6795ab99246937c629cd40880f51efa2012d", - "bytes": 9312 + "sha256": "285c5275b83f4fdf62882e6493d4a24f17c4187971b74def041af32f620fda37", + "bytes": 9550 }, { "path": "src/thoth/application/services/authorization_currentness.py", - "sha256": "314909284bdee460f10b110cacb2697c820c8f42d615345f42c488a85ee2b9cc", - "bytes": 6218 + "sha256": "a969f0f27d023c1148fe19a8e6b9eda70c31b028dde33b60738d7cac25173c6b", + "bytes": 6791 }, { "path": "src/thoth/application/services/authorization_prepare.py", @@ -6291,6 +6301,11 @@ "sha256": "3af3e30c62887ecb7330a5bdbeb5f84133b4f38da0323b08ad7505d816dc25a0", "bytes": 2982 }, + { + "path": "src/thoth/domain/authorization_errors.py", + "sha256": "bba7b32331d446b46a8909b958028b3cc9acbd817880b82f3f5678e40db3c523", + "bytes": 507 + }, { "path": "src/thoth/domain/base.py", "sha256": "65387dc8a7fe6e4b15d44481409a5c5d4cc623347e78f973a9301499e4a13fe4", @@ -6878,8 +6893,8 @@ }, { "path": "src/thoth/ports/action.py", - "sha256": "8154012bbe2db3622c56d534a927032f13431bde2e7b26752afc8b8fd6dcd631", - "bytes": 1634 + "sha256": "164445fc73cb715020547933b2bb00aaadf49151cdf388de93653b90f6c4c87f", + "bytes": 1765 }, { "path": "src/thoth/ports/artifact_ledger.py", @@ -7668,28 +7683,28 @@ }, { "path": "tests/contract/test_async_http.py", - "sha256": "fa8526a8d4fab52d656ca3ddbdfb36df69652106c5740b1aba18aedd3564e4ec", - "bytes": 9984 + "sha256": "163017d0ee1815479271c8598dae694793097f9634ee7a068b6094cc46c5f32b", + "bytes": 10048 }, { "path": "tests/contract/test_desktop_web_mount.py", - "sha256": "ef48ef6f97b64fd5374c686d77fd03edc1ddcb4d7b4a05a5c3d23a7e216c432a", - "bytes": 1429 + "sha256": "0a712c6ed8082bb5599f8c32b2f23f5a3577c1adb5a0d413f23d4acf2ba7ea8e", + "bytes": 1463 }, { "path": "tests/contract/test_file_upload.py", - "sha256": "d81442178296d5fe28fac36a02cef4c96c2a14cfea5e3fc8b46781706ad21ebe", - "bytes": 5427 + "sha256": "5ea6cdb29e76b7793e1650789e7224f69689a85b7d36134f9086e1201c488a1e", + "bytes": 5487 }, { "path": "tests/contract/test_http_async_route_contract.py", - "sha256": "be022fcde55841cc8bd6ad34201a8dda7de07717193179eb6d94c3b58fea76c1", - "bytes": 15473 + "sha256": "fde31969e4e1a5ea1af2dd0f5ed0271f7627b52a88e58977d8f7aab4f9b9d079", + "bytes": 15643 }, { "path": "tests/contract/test_http_file_staging_contract.py", - "sha256": "59f1c95c8b2350cd8e7e9d4089d7932bdc8081459d40167b561bd0ebcbe14839", - "bytes": 13561 + "sha256": "19c6dfafd267700dfbaac8ac60354bc778aba097f075df910820fbcdd96b8e39", + "bytes": 13581 }, { "path": "tests/contract/test_http_internal_route_contract.py", @@ -7703,8 +7718,13 @@ }, { "path": "tests/contract/test_local_loopback_identity.py", - "sha256": "882fbfce49d336f36135d8b280488c4fb5e84358d9945ca77bda4c6e55c9a0ae", - "bytes": 3120 + "sha256": "598dc05121ed50e50b8a7a0afe3670953484517e06e730a53ea7fbb68673e89b", + "bytes": 3184 + }, + { + "path": "tests/contract/test_local_request_guard.py", + "sha256": "7c24684d360a6f758182b2bb5820043e320ba2e34c9cb1fa0b1717f1a6255633", + "bytes": 4574 }, { "path": "tests/contract/test_method_registry.py", @@ -7723,8 +7743,8 @@ }, { "path": "tests/contract/test_transports.py", - "sha256": "cb060594256bf0e2c6b7d4090dd653b64f5bba528a3c0f1510f3d0b0ad539b1a", - "bytes": 1731 + "sha256": "2c5cba56f270fdca92adfd3387940e88d92782381c36dae5d60fc053d0b4c6b8", + "bytes": 1741 }, { "path": "tests/fixtures/action_response_golden.json", @@ -8048,8 +8068,8 @@ }, { "path": "tests/integration/resource_scope_helpers.py", - "sha256": "ce99ea057790d2a20ee14f5b8640f570c2712211015f4df74a5588ae0a86d984", - "bytes": 10294 + "sha256": "8b8453c4ca92859e0eaca5cb5998791e2e635103bb54fe69bc7e3b0654d8a765", + "bytes": 10314 }, { "path": "tests/integration/schema_contract_probe.py", @@ -8178,8 +8198,8 @@ }, { "path": "tests/integration/test_a06_memory_storage_boundary.py", - "sha256": "94ac0e2ed1fe062e0669ba5e06fbb89a71bb74435f8e5d1b540659da4c4a077b", - "bytes": 21155 + "sha256": "af07d3290a560d687b076fb608d25cc385a311339d09ef6a2c831bc8b4b677d4", + "bytes": 21165 }, { "path": "tests/integration/test_a06_semantic_memory_review.py", @@ -8213,8 +8233,8 @@ }, { "path": "tests/integration/test_a08_storage_authority.py", - "sha256": "b8dfd177c5e239e7fcb95868ac42ed58ee5f31f451a9051caa257bbc061160b3", - "bytes": 36449 + "sha256": "1a3c842f54e152ddc28b953a9e64bbcf0e54c3aeb60db1ce52d6b7b76745fcdf", + "bytes": 36465 }, { "path": "tests/integration/test_a08_workflow_candidate_canary.py", @@ -8288,13 +8308,13 @@ }, { "path": "tests/integration/test_a13_authenticated_multiplayer.py", - "sha256": "5cd1bdeb3721d97638e70337f840a5d53459a36924de3b2acd04d66df4190f7a", - "bytes": 26313 + "sha256": "4c55a647577217d48d5243c5add782811bbe873252af4c750556a9a8f26c726d", + "bytes": 26377 }, { "path": "tests/integration/test_a13_operation_scope.py", - "sha256": "5346ee1bf387b4b9e644ebdef1b97d17a26ad5e83df7ff41c81b081d7c5e7941", - "bytes": 13502 + "sha256": "e5ae9a29f5deb35b3c49103a575416674ecf740e4f796fd7cf41358d35f14e1f", + "bytes": 13566 }, { "path": "tests/integration/test_a13_project_atomicity.py", @@ -8303,8 +8323,8 @@ }, { "path": "tests/integration/test_a13_security_boundaries.py", - "sha256": "16fb26b6bd144b640e3436591dabcf65a6925c7c7db5697869c76398badd0d9c", - "bytes": 10306 + "sha256": "e245293ecb4c541cd96be2219ce815e989e7cd130b62ab0d922ad3e1fbf3f76f", + "bytes": 10370 }, { "path": "tests/integration/test_acquisition_io_transaction_boundary.py", @@ -8351,6 +8371,11 @@ "sha256": "6742bf34af49f3c670e4974231c7071cdae3cfde60c1b2776b12e8164db2fcdf", "bytes": 12846 }, + { + "path": "tests/integration/test_approval_clock_rollback.py", + "sha256": "186f2abacc70a30903721ed24405949f530955c87dd025e94d4e27aba07dda21", + "bytes": 11887 + }, { "path": "tests/integration/test_architecture_cleanup_projectpack_migration.py", "sha256": "a4263083dbb437ec41f5af63bec4782b85dddf9a7c9ffa5b62722e832ec956a1", @@ -8361,6 +8386,11 @@ "sha256": "6512f9f74787e334e9acf8b3ee0d966efb307f5e7784e2fb4ece21812f01b5fe", "bytes": 7734 }, + { + "path": "tests/integration/test_authorization_consumption_guards.py", + "sha256": "f97efd5bd971e918087b0340536791ab6e4460bc6f7142ee26163911af436527", + "bytes": 7462 + }, { "path": "tests/integration/test_automatic_behavior_candidate.py", "sha256": "2b81f0c862aa7fd0bf134b2f901634c94ecb996f2b7eec4d9b61404891fb380f", @@ -8418,8 +8448,8 @@ }, { "path": "tests/integration/test_codex_auth_nonblocking.py", - "sha256": "daa480d1c52fbeadd2b2caee6bd7d56237d17b0b472513cb0b4ae7bca0602407", - "bytes": 6710 + "sha256": "23dde076406dad87aebf142cf3e1c38a0d0a51c263ec743b5573cb9091bb98b0", + "bytes": 6655 }, { "path": "tests/integration/test_confirmed_findings_result.py", @@ -8438,8 +8468,8 @@ }, { "path": "tests/integration/test_counter_cleanup_cancel_preserves_prior.py", - "sha256": "3f0bcaad2f73a291fc3f21e9bc2b0d8e81ec08d393c154a4dff0300e9a1f654b", - "bytes": 6509 + "sha256": "6b91a1884d2ba1f058fbe3a986ced8860e2d877494cdf4a3bb513fe66cc0f619", + "bytes": 6519 }, { "path": "tests/integration/test_counter_clock_sequence.py", @@ -8488,8 +8518,8 @@ }, { "path": "tests/integration/test_decision_object_restore_fail_closed.py", - "sha256": "ab37bce7fffcbf1d6570a33ddbeb27caab06b293ce4cd49cf7e3d0ba58697bbc", - "bytes": 14951 + "sha256": "8213193fd89a857c2b095eeaf38cbc984efa4cf3361ca9454cbc6f88501d71b3", + "bytes": 14961 }, { "path": "tests/integration/test_discovery_cleanup_cancel.py", @@ -8701,6 +8731,11 @@ "sha256": "af45929a586dbbd649754cb1d98e1f1cbf23d6490354759078099c0747be1c51", "bytes": 13507 }, + { + "path": "tests/integration/test_local_http_origin_baseline.py", + "sha256": "42cf5778c0a8ddb7892d4241c9f3ce9c738f6cb6b82a7550a0415c971048308c", + "bytes": 12691 + }, { "path": "tests/integration/test_local_workspace_restart.py", "sha256": "4490d53dc07f9078c23d48a1bb81934d0ab3871588654462baa4de307c1aeb99", @@ -8708,8 +8743,8 @@ }, { "path": "tests/integration/test_m2_counter_cancel_boundary.py", - "sha256": "b4896eef4cb2ff27fa5e57a28dde58e275463f5191a2a1cb60dc92ef9763c85c", - "bytes": 9435 + "sha256": "6a2d0d100925a0f23054980e8623da3329e83746467f221b030ac670c796c380", + "bytes": 9445 }, { "path": "tests/integration/test_m2_counter_publication_boundaries.py", @@ -8881,6 +8916,11 @@ "sha256": "10eb73a68f02d0dc196a42f77e1f75053eaed787abc1dafd8a024fdb398b2c58", "bytes": 1039 }, + { + "path": "tests/integration/test_migration_logging.py", + "sha256": "031966cbe2165602bdbbe20a3e185c17142bfc2a76229c7a2f97bddf2ea6a3d8", + "bytes": 1165 + }, { "path": "tests/integration/test_migrations.py", "sha256": "161c0915421214899582f01c1880f482fa40e9a26a7209c7a2786142ced5e89d", @@ -9018,8 +9058,8 @@ }, { "path": "tests/integration/test_prepare_synthetic_radar_demo.py", - "sha256": "727c9dd658b3bd54f7dcc1d6fb212268173db68ec9b1cc7a309900a1f473a7ad", - "bytes": 7774 + "sha256": "ab4cd8bd35adcc5b16f7738c4ec94cb411cc22f01bfdd8cba799ae6d458c56e2", + "bytes": 7808 }, { "path": "tests/integration/test_profile_cross_domain_overfit.py", @@ -9076,6 +9116,11 @@ "sha256": "dc229b7a34885220186e823cc9879676187dc5db2a9c6a470487812913caf256", "bytes": 8251 }, + { + "path": "tests/integration/test_public_browser_cleanup.py", + "sha256": "995080afcd65117afd12580ebf30653811f852cb229da2c05a65146434ab1369", + "bytes": 6577 + }, { "path": "tests/integration/test_public_browser_navigation.py", "sha256": "d012cee2595cf952a24991aea3d46ded869d954b19ebb3d7c54eea01be3f5f1b", @@ -9263,8 +9308,8 @@ }, { "path": "tests/integration/test_research_input_queue.py", - "sha256": "5124f1ae397930f4c42a2f7a1b90cbb8d89f29479759ac0d8615cdc53d4eff74", - "bytes": 38082 + "sha256": "ee01159936e592cea2db73e5119aa498f415fdcee2c3bd9a4a80b987ccc62ec2", + "bytes": 38078 }, { "path": "tests/integration/test_research_lineage_publication.py", @@ -9443,8 +9488,8 @@ }, { "path": "tests/integration/test_restore_access.py", - "sha256": "b5d55e85127252afd82eb1be08149886d2aab47d42dcc0164bf40f23688f8a96", - "bytes": 4823 + "sha256": "ad062596a07690f567748cf63b4b230a04bc92437f974e540e3820a650484a65", + "bytes": 4833 }, { "path": "tests/integration/test_restore_apply_atomicity.py", @@ -9618,8 +9663,8 @@ }, { "path": "tests/integration/test_source_cleanup_cancel_boundary.py", - "sha256": "4631c061d12f4b6b004adbfc68655ac2bb6aa1ca46a97320383ca5ca5ec60233", - "bytes": 7855 + "sha256": "36f8c9569bb9c197a48aace5c4a49b0bfe1e4c4827b3b408051dfc49f0839b83", + "bytes": 7865 }, { "path": "tests/integration/test_source_list_cost.py", @@ -9778,18 +9823,18 @@ }, { "path": "tests/integration/test_workspace_runtime_exclusivity_baseline.py", - "sha256": "11a3ea83a1afa6d44997238b13956e36f36073bdd9c6af6939abf1f9213c6c5e", - "bytes": 7958 + "sha256": "4d8930545fb47e2a62f2ed7255c280ef82c58de931abddbc9570548b1ee11271", + "bytes": 7990 }, { "path": "tests/integration/test_workspace_runtime_lock.py", - "sha256": "dfc9c375c2e2eff25781411d6e2d1b5ecc16249ab4961c12e47aef98837ee5f4", - "bytes": 13669 + "sha256": "e1f2752589771c225cf0a0582137d13813aa6d3e5ad907953ed871413339c62e", + "bytes": 13701 }, { "path": "tests/integration/test_xai_workspace_route.py", - "sha256": "1aa50ddba2fde60b5c782eb7fab14820bf24041fbed75b7e145ba63d316b8280", - "bytes": 21938 + "sha256": "75b88f441ec7efc0358d4b2fd9868ce45091fa59d8178144093b0793d5950788", + "bytes": 22143 }, { "path": "tests/integration/trace_demo_helpers.py", @@ -10696,6 +10741,11 @@ "sha256": "0f69727a0c108911b5bd1ece7c141cdb8f8e826c9e5f900abd810c0542216896", "bytes": 14549 }, + { + "path": "tests/unit/services/test_authorization_clock_currentness.py", + "sha256": "6b9c63ca0f08395897a0244ee60e012fc7ad94fb3f99ef94bae0d377a1bf1ec8", + "bytes": 4277 + }, { "path": "tests/unit/services/test_authorization_currentness.py", "sha256": "b8e38880709329ad4fe4618305b07e7377fe03bc30659b303b333367a3b330b3", @@ -10911,6 +10961,11 @@ "sha256": "059aa3d2dfde3d72baddc26b7a31b0295f4d26004c662ec946244c353dc0e011", "bytes": 2329 }, + { + "path": "tests/unit/test_anonymous_browser_lifecycle.py", + "sha256": "0536771653420b5493b4efc8e02b1d11cd0a1068f89f16989068254f24193a18", + "bytes": 7348 + }, { "path": "tests/unit/test_cli_auth_presentation_contract.py", "sha256": "47a5514c042bc38d98cca4a38ca0600ebbae46c4dd4832efbebcefe5ea9ebb89", @@ -11036,6 +11091,11 @@ "sha256": "a20593ab7789905988c265ca75f5297575d57e39f3d5c81ee8980bb110dd1ddb", "bytes": 3333 }, + { + "path": "tests/unit/test_local_request_guard_unit.py", + "sha256": "4e70edd31bc1a7316799a8fa96a6050185b6cc8f5605ea6d570273d4b96b63a1", + "bytes": 7865 + }, { "path": "tests/unit/test_memory_eval_families.py", "sha256": "263d5b0c083e59d6ec821b17d16c2a3421c35ca7d2bf9e45a6548e0065fbeb19", diff --git a/docs/KNOWN_LIMITATIONS.ko.md b/docs/KNOWN_LIMITATIONS.ko.md index 6344457..f1b2ea6 100644 --- a/docs/KNOWN_LIMITATIONS.ko.md +++ b/docs/KNOWN_LIMITATIONS.ko.md @@ -7,6 +7,10 @@ THOTH는 실험 단계의 소스 미리보기입니다. 이 문서는 저장된 English: [KNOWN_LIMITATIONS.md](KNOWN_LIMITATIONS.md) +## 로컬 모드는 내 컴퓨터의 사용자를 신뢰합니다 + +로컬 모드는 한 사람이 사용하는 환경을 전제로, 같은 컴퓨터의 프로그램을 별도 인증 없이 신뢰합니다. 서버는 로컬 Host 이름을 확인하고, 상태를 바꿀 수 있는 요청에 Origin이 있으면 로컬 HTTP 출처만 허용합니다. RPC와 취소 요청은 JSON 본문이어야 합니다. 다른 Host·Origin이나 평문 RPC 요청은 거절하며 내장 화면, Vite 개발 프록시, 로컬 JSON 클라이언트는 그대로 사용할 수 있습니다. 이 검사는 인증을 대신하지 않습니다. 같은 컴퓨터의 다른 프로그램은 Origin을 생략하고 허용된 헤더를 보낼 수 있으므로 막지 못합니다. 심사용(hosted review) 모드는 별도 접근 통제를 사용합니다. + ## 작업 공간마다 THOTH는 하나만 실행됩니다 작업 공간 하나는 한 번에 THOTH 프로세스 하나만 사용합니다. 같은 작업 공간에서 두 번째를 diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index e4781a1..c70e96a 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -8,6 +8,10 @@ verification state is in [VERIFICATION.md](VERIFICATION.md). 한국어 안내: [KNOWN_LIMITATIONS.ko.md](KNOWN_LIMITATIONS.ko.md) +## Local access trusts the user of this computer + +Local mode assumes one user and trusts programs on the same computer without authenticating them. The server checks local Host names and, for requests that can change state, accepts only local HTTP Origins when an Origin is present; RPC and cancellation requests also require JSON. This rejects requests with foreign Host/Origin values and plain-text RPC bodies while preserving the built-in UI, the Vite development proxy and local JSON clients. These checks are not authentication: another program on this computer can omit Origin and send allowed headers. Hosted review has separate access controls. + ## One THOTH process per workspace is enforced A workspace is open in one THOTH process at a time. Starting a second one on the same workspace diff --git a/migrations/env.py b/migrations/env.py index c3b5214..f197ebc 100644 --- a/migrations/env.py +++ b/migrations/env.py @@ -15,7 +15,7 @@ config = context.config if config.config_file_name is not None: - fileConfig(config.config_file_name) + fileConfig(config.config_file_name, disable_existing_loggers=False) override_url = os.getenv("THOTH_ALEMBIC_URL") if override_url: diff --git a/src/thoth/adapters/connectors/anonymous_browser.py b/src/thoth/adapters/connectors/anonymous_browser.py index c4a40e0..6715f2e 100644 --- a/src/thoth/adapters/connectors/anonymous_browser.py +++ b/src/thoth/adapters/connectors/anonymous_browser.py @@ -1,8 +1,13 @@ """Fresh browser context; all page network reads are fulfilled by the bounded broker.""" +from __future__ import annotations + import asyncio import json +import sys +from typing import TYPE_CHECKING +from thoth.adapters.connectors.anonymous_browser_lifecycle import AnonymousBrowserLifecycle from thoth.adapters.connectors.browser_document_capture import ( capture_document, create_document_identity, @@ -12,6 +17,9 @@ from thoth.domain.web_acquisition import WebNavigationHop, WebPage from thoth.ports.web_reader import PublicReaderPort +if TYPE_CHECKING: + from playwright.async_api import Browser + class AnonymousChromiumReader: def __init__(self, reader: PublicReaderPort, *, timeout_seconds: float = 20) -> None: @@ -19,174 +27,167 @@ def __init__(self, reader: PublicReaderPort, *, timeout_seconds: float = 20) -> self.timeout_seconds = timeout_seconds async def render(self, page: WebPage, *, max_bytes: int, timeout: float) -> WebPage: - return await asyncio.wait_for( - self._render(page, max_bytes=max_bytes, timeout=min(timeout, self.timeout_seconds)), - min(timeout, self.timeout_seconds) + 2, - ) - - async def _render(self, page: WebPage, *, max_bytes: int, timeout: float) -> WebPage: + lifetime = AnonymousBrowserLifecycle() + bounded = min(timeout, self.timeout_seconds) try: - from playwright.async_api import Route, async_playwright - except ImportError as exc: - raise ConnectorFailure( - ConnectorErrorCode.DRIVER_ERROR, "ANONYMOUS_BROWSER_NOT_INSTALLED" - ) from exc - async with async_playwright() as driver: - browser = await driver.chromium.launch( - headless=True, - timeout=timeout * 1000, - chromium_sandbox=True, - args=["--disable-background-networking", "--disable-quic", "--no-proxy-server"], + async with asyncio.timeout(bounded + 2): + browser = await lifetime.launch(bounded) + return await self._render(page, browser, max_bytes=max_bytes, timeout=bounded) + finally: + primary = sys.exception() + try: + await lifetime.close() + except ConnectorFailure: + if primary is None: + raise + + async def _render( + self, page: WebPage, browser: Browser, *, max_bytes: int, timeout: float + ) -> WebPage: + from playwright.async_api import Route + + context = await browser.new_context(accept_downloads=False, service_workers="block") + remaining, calls = max_bytes * 2, 0 + failures: list[str] = [] + documents: dict[str, WebPage] = {} + redirect_cache: dict[str, WebPage] = {} + pending_documents: set[str] = set() + navigation = list(page.navigation) + prior = page.requested_uri + for uri in page.redirects: + navigation.append( + WebNavigationHop( + from_uri=prior, + to_uri=uri, + kind="HTTP_REDIRECT", + retrieved_at=page.retrieved_at, + document_sha256=content_digest(page.raw) if uri == page.final_uri else None, + ) ) + prior = uri + + async def route(request: Route) -> None: + nonlocal remaining, calls + calls += 1 + if request.request.method != "GET" or calls > 32 or remaining <= 0: + failures.append("BROWSER_REQUEST_LIMIT") + await request.abort() + return + if request.request.resource_type not in { + "document", + "script", + "stylesheet", + "xhr", + "fetch", + }: + await request.abort() + return try: - context = await browser.new_context(accept_downloads=False, service_workers="block") - remaining, calls = max_bytes * 2, 0 - failures: list[str] = [] - documents: dict[str, WebPage] = {} - redirect_cache: dict[str, WebPage] = {} - pending_documents: set[str] = set() - navigation = list(page.navigation) - prior = page.requested_uri - for uri in page.redirects: - navigation.append( - WebNavigationHop( - from_uri=prior, - to_uri=uri, - kind="HTTP_REDIRECT", - retrieved_at=page.retrieved_at, - document_sha256=content_digest(page.raw) - if uri == page.final_uri - else None, - ) + url = request.request.url + cached = redirect_cache.pop(url, None) + content = cached or ( + page + if url == page.final_uri and calls == 1 + else await self.reader.read( + request.request.url, + max_bytes=min(max_bytes, remaining), + timeout=timeout, ) - prior = uri - - async def route(request: Route) -> None: - nonlocal remaining, calls - calls += 1 - if request.request.method != "GET" or calls > 32 or remaining <= 0: - failures.append("BROWSER_REQUEST_LIMIT") - await request.abort() - return - if request.request.resource_type not in { - "document", - "script", - "stylesheet", - "xhr", - "fetch", - }: - await request.abort() - return - try: - url = request.request.url - cached = redirect_cache.pop(url, None) - content = cached or ( - page - if url == page.final_uri and calls == 1 - else await self.reader.read( - request.request.url, - max_bytes=min(max_bytes, remaining), - timeout=timeout, + ) + if cached is None: + remaining -= len(content.raw) + if remaining < 0 or len(content.raw) > max_bytes: + raise ValueError("BROWSER_RESPONSE_LIMIT") + main_document = ( + request.request.is_navigation_request() + and request.request.frame == tab.main_frame + ) + if content.final_uri != url: + # The broker followed validated redirects. Make Chromium commit + # that URI, rather than fulfilling other-document bytes at url. + redirect_cache[content.final_uri] = content + if not main_document: + raise ValueError("BROWSER_SUBRESOURCE_REDIRECT_UNSUPPORTED") + if main_document: + prior_uri = url + for uri in content.redirects or (content.final_uri,): + navigation.append( + WebNavigationHop( + from_uri=prior_uri, + to_uri=uri, + kind="HTTP_REDIRECT", + retrieved_at=content.retrieved_at, + document_sha256=( + content_digest(content.raw) + if uri == content.final_uri + else None + ), + ) ) - ) - if cached is None: - remaining -= len(content.raw) - if remaining < 0 or len(content.raw) > max_bytes: - raise ValueError("BROWSER_RESPONSE_LIMIT") - main_document = ( - request.request.is_navigation_request() - and request.request.frame == tab.main_frame - ) - if content.final_uri != url: - # The broker followed validated redirects. Make Chromium commit - # that URI, rather than fulfilling other-document bytes at url. - redirect_cache[content.final_uri] = content - if not main_document: - raise ValueError("BROWSER_SUBRESOURCE_REDIRECT_UNSUPPORTED") - if main_document: - prior_uri = url - for uri in content.redirects or (content.final_uri,): - navigation.append( - WebNavigationHop( - from_uri=prior_uri, - to_uri=uri, - kind="HTTP_REDIRECT", - retrieved_at=content.retrieved_at, - document_sha256=( - content_digest(content.raw) - if uri == content.final_uri - else None - ), - ) - ) - prior_uri = uri - # A fulfilled HTTP 302 may bypass Playwright routing on - # its redirect request. An explicit document navigation - # is intercepted again, using the broker-validated cache. - shim = ( - "" - ).encode("utf-8") - remaining -= len(shim) - if remaining < 0: - raise ValueError("BROWSER_RESPONSE_LIMIT") - await request.fulfill(status=200, content_type="text/html", body=shim) - return - if main_document: - documents[url] = content - pending_documents.add(url) - await request.fulfill( - status=200, content_type=content.media_type, body=content.raw - ) - except Exception: - failures.append("BROWSER_RESOURCE_READ_FAILED") - await request.abort() + prior_uri = uri + # A fulfilled HTTP 302 may bypass Playwright routing on + # its redirect request. An explicit document navigation + # is intercepted again, using the broker-validated cache. + shim = ( + "" + ).encode("utf-8") + remaining -= len(shim) + if remaining < 0: + raise ValueError("BROWSER_RESPONSE_LIMIT") + await request.fulfill(status=200, content_type="text/html", body=shim) + return + if main_document: + documents[url] = content + pending_documents.add(url) + await request.fulfill(status=200, content_type=content.media_type, body=content.raw) + except Exception: + failures.append("BROWSER_RESOURCE_READ_FAILED") + await request.abort() - await context.route("**/*", route) - await context.route_web_socket("**/*", lambda socket: socket.close()) - tab = await context.new_page() - tab.set_default_timeout(timeout * 1000) + await context.route("**/*", route) + await context.route_web_socket("**/*", lambda socket: socket.close()) + tab = await context.new_page() + tab.set_default_timeout(timeout * 1000) - native_identity = await create_document_identity( - context, tab, documents, pending_documents, navigation, page.final_uri - ) + native_identity = await create_document_identity( + context, tab, documents, pending_documents, navigation, page.final_uri + ) - raw, document_uri, location = await asyncio.wait_for( - capture_document( - tab, - page.final_uri, - documents, - native_identity, - max_bytes, - timeout, - ), - timeout, - ) - if failures: - raise ConnectorFailure(ConnectorErrorCode.PARTIAL_FETCH, failures[0]) - if len(raw) > max_bytes: - raise ConnectorFailure( - ConnectorErrorCode.CONTENT_LIMIT_EXCEEDED, "BROWSER_DOM_TOO_LARGE" - ) - if location != document_uri: - navigation.append( - WebNavigationHop( - from_uri=document_uri, - to_uri=location, - kind="HISTORY_LOCATION", - retrieved_at=utc_now(), - ) - ) - return WebPage( - requested_uri=page.requested_uri, - final_uri=document_uri, - media_type="text/html", - raw=raw, + raw, document_uri, location = await asyncio.wait_for( + capture_document( + tab, + page.final_uri, + documents, + native_identity, + max_bytes, + timeout, + ), + timeout, + ) + if failures: + raise ConnectorFailure(ConnectorErrorCode.PARTIAL_FETCH, failures[0]) + if len(raw) > max_bytes: + raise ConnectorFailure( + ConnectorErrorCode.CONTENT_LIMIT_EXCEEDED, "BROWSER_DOM_TOO_LARGE" + ) + if location != document_uri: + navigation.append( + WebNavigationHop( + from_uri=document_uri, + to_uri=location, + kind="HISTORY_LOCATION", retrieved_at=utc_now(), - redirects=tuple(h.to_uri for h in navigation if h.kind == "HTTP_REDIRECT"), - navigation=tuple(navigation), - dom_location_at_capture=location, ) - finally: - await asyncio.wait_for(browser.close(), 2) + ) + return WebPage( + requested_uri=page.requested_uri, + final_uri=document_uri, + media_type="text/html", + raw=raw, + retrieved_at=utc_now(), + redirects=tuple(h.to_uri for h in navigation if h.kind == "HTTP_REDIRECT"), + navigation=tuple(navigation), + dom_location_at_capture=location, + ) diff --git a/src/thoth/adapters/connectors/anonymous_browser_lifecycle.py b/src/thoth/adapters/connectors/anonymous_browser_lifecycle.py new file mode 100644 index 0000000..3bb38b2 --- /dev/null +++ b/src/thoth/adapters/connectors/anonymous_browser_lifecycle.py @@ -0,0 +1,99 @@ +"""Own one render's browser and driver, independently of its reading deadline.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING + +from thoth.domain.connectors import ConnectorErrorCode, ConnectorFailure + +if TYPE_CHECKING: + from playwright.async_api import Browser + +_log = logging.getLogger(__name__) +_pending: set[asyncio.Task[None]] = set() + + +def _owned_task(action: Awaitable[None]) -> asyncio.Task[None]: + async def invoke() -> None: + await action + + task = asyncio.create_task(invoke()) + _pending.add(task) + + def finished(done: asyncio.Task[None]) -> None: + _pending.discard(done) + if not done.cancelled(): + done.exception() + + task.add_done_callback(finished) + return task + + +class AnonymousBrowserLifecycle: + def __init__(self) -> None: + self.browser: Browser | None = None + self._stop_driver: Callable[[], Awaitable[None]] | None = None + + async def launch(self, timeout: float) -> Browser: + try: + from playwright.async_api import async_playwright + except ImportError as exc: + raise ConnectorFailure( + ConnectorErrorCode.DRIVER_ERROR, "ANONYMOUS_BROWSER_NOT_INSTALLED" + ) from exc + manager = async_playwright() + self._stop_driver = manager.__aexit__ + driver = await manager.start() + self.browser = await driver.chromium.launch( + headless=True, + timeout=timeout * 1000, + chromium_sandbox=True, + args=["--disable-background-networking", "--disable-quic", "--no-proxy-server"], + ) + return self.browser + + async def close(self) -> None: + """Finish owned cleanup before propagating any caller cancellation.""" + cleanup = _owned_task(self._cleanup()) + cancelled = False + while True: + try: + await asyncio.shield(cleanup) + break + except asyncio.CancelledError: + if cleanup.cancelled(): + raise + cancelled = True + except Exception: + if cancelled: + raise asyncio.CancelledError from None + raise + if cancelled: + raise asyncio.CancelledError + + async def _cleanup(self) -> None: + if self._stop_driver is None: + return + closing = None if self.browser is None else _owned_task(self.browser.close()) + if closing is not None: + done, _ = await asyncio.wait({closing}, timeout=2) + if not done: + _log.warning("ANONYMOUS_BROWSER_CLOSE_DELAYED") + elif closing.cancelled() or closing.exception() is not None: + _log.warning("ANONYMOUS_BROWSER_CLOSE_FAILED") + # Closing this call's driver pipe re-enters SDK browser cleanup and forces + # a browser whose graceful close is already in progress to exit. + stopping = _owned_task(self._stop_driver()) + done, _ = await asyncio.wait({stopping}, timeout=2) + confirmed = bool(done) and not stopping.cancelled() and stopping.exception() is None + for task in (closing, stopping): + if task is not None and not task.done(): + task.cancel() + if not confirmed: + _log.error("ANONYMOUS_BROWSER_CLEANUP_UNCONFIRMED") + raise ConnectorFailure( + ConnectorErrorCode.DRIVER_ERROR, "ANONYMOUS_BROWSER_CLEANUP_UNCONFIRMED" + ) diff --git a/src/thoth/adapters/http/app.py b/src/thoth/adapters/http/app.py index dba5627..a822986 100644 --- a/src/thoth/adapters/http/app.py +++ b/src/thoth/adapters/http/app.py @@ -29,6 +29,7 @@ is_hosted_credential_method, ) from thoth.adapters.http.internal_routes import mount_internal_routes +from thoth.adapters.http.local_request_guard import LocalRequestGuard from thoth.adapters.storage.workspace_owner import WorkspaceInUse from thoth.apps.hosted_review_dispatch import hosted_dispatch from thoth.apps.hosted_review_snapshot import ( @@ -217,6 +218,8 @@ async def lifespan(application: FastAPI) -> AsyncGenerator[None]: version=__version__, lifespan=lifespan, ) + if not hosted_mode(): + application.add_middleware(LocalRequestGuard) application.state.bus = bus application.state.runtime = None application.state.session_runtimes = {} @@ -309,7 +312,6 @@ async def stage_file(request: Request, file: UploadFile = File(...)) -> dict[str return application - def _mount_async_rpc_route( application: FastAPI, auth: HttpAuthenticationPort | None, local_actor_id: str ) -> None: @@ -337,13 +339,10 @@ def _mount_internal_routes(application: FastAPI) -> None: _resolve_bus_for_request=lambda app, request: _resolve_bus_for_request(app, request), ) + def web_ui_dist() -> Path | None: raw = os.environ.get("THOTH_WEB_DIST") - dist = ( - Path(raw) - if raw - else Path(__file__).resolve().parents[4] / "apps" / "web" / "dist" - ) + dist = Path(raw) if raw else Path(__file__).resolve().parents[4] / "apps" / "web" / "dist" if (dist / "index.html").is_file(): return dist return None diff --git a/src/thoth/adapters/http/local_request_guard.py b/src/thoth/adapters/http/local_request_guard.py new file mode 100644 index 0000000..8b89928 --- /dev/null +++ b/src/thoth/adapters/http/local_request_guard.py @@ -0,0 +1,125 @@ +"""Reject nonlocal request origins before body, runtime or storage access.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Literal +from urllib.parse import urlsplit + +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Receive, Scope, Send + +from thoth.adapters.http.hosted_review import hosted_mode + +_NAMES = frozenset({"127.0.0.1", "localhost", "::1"}) +_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) +_JSON_PATHS = frozenset({"/rpc", "/rpc/query", "/rpc/async"}) +_CANCEL_PATH = re.compile(r"/operations/[^/]+/cancel") + + +@dataclass(frozen=True) +class LocalRequestRejection: + status: int + code: Literal["LOCAL_HOST_REJECTED", "LOCAL_ORIGIN_REJECTED", "JSON_CONTENT_TYPE_REQUIRED"] + message_ko: str + message_en: str + + def payload(self) -> dict[str, str]: + return {"code": self.code, "message_ko": self.message_ko, "message_en": self.message_en} + + +_HOST = LocalRequestRejection(403, "LOCAL_HOST_REJECTED", "호스트 거부", "Host rejected") +_ORIGIN = LocalRequestRejection(403, "LOCAL_ORIGIN_REJECTED", "출처 거부", "Origin rejected") +_JSON = LocalRequestRejection(415, "JSON_CONTENT_TYPE_REQUIRED", "JSON 필요", "JSON required") + + +def _header_values(headers: Sequence[tuple[bytes, bytes]], name: bytes) -> list[str]: + return [value.decode("latin-1") for key, value in headers if key.lower() == name] + + +def _local_authority(value: str, *, origin: bool) -> bool: + if not value or any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in value): + return False + try: + parsed = urlsplit(value if origin else "//" + value) + if origin and parsed.scheme.lower() != "http": + return False + if ( + parsed.hostname not in _NAMES + or parsed.username is not None + or parsed.password is not None + or parsed.path + or parsed.query + or parsed.fragment + or "?" in value + or "#" in value + or "\\" in value + ): + return False + # Validate port syntax without restricting the server or proxy's port. + _ = parsed.port + authority = parsed.netloc + if authority.endswith(":"): + return False + return parsed.hostname != "::1" or authority == "[::1]" or authority.startswith("[::1]:") + except ValueError: + return False + + +def check_local_request( + method: str, + path: str, + headers: Sequence[tuple[bytes, bytes]], + *, + websocket: bool = False, +) -> LocalRequestRejection | None: + hosts = _header_values(headers, b"host") + if len(hosts) != 1 or not _local_authority(hosts[0], origin=False): + return _HOST + if not websocket and method.upper() in _SAFE_METHODS: + return None + origins = _header_values(headers, b"origin") + if origins and (len(origins) != 1 or not _local_authority(origins[0], origin=True)): + return _ORIGIN + normalized = path.rstrip("/") + if not websocket and (normalized in _JSON_PATHS or _CANCEL_PATH.fullmatch(normalized)): + types = _header_values(headers, b"content-type") + if len(types) != 1 or types[0].split(";", 1)[0].strip().lower() != "application/json": + return _JSON + return None + + +class LocalRequestGuard: + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + kind = scope["type"] + if kind not in {"http", "websocket"} or hosted_mode(): + await self.app(scope, receive, send) + return + rejection = check_local_request( + scope.get("method", "GET"), + scope["path"], + scope["headers"], + websocket=kind == "websocket", + ) + if rejection is None: + await self.app(scope, receive, send) + elif kind == "websocket": + await send( + { + "type": "websocket.close", + "code": 1008, + "reason": json.dumps( + rejection.payload(), ensure_ascii=False, separators=(",", ":") + ), + } + ) + else: + await JSONResponse(rejection.payload(), status_code=rejection.status)( + scope, receive, send + ) diff --git a/src/thoth/adapters/storage/action.py b/src/thoth/adapters/storage/action.py index b2ca501..c576f0d 100644 --- a/src/thoth/adapters/storage/action.py +++ b/src/thoth/adapters/storage/action.py @@ -292,10 +292,7 @@ def read_authorization( action_authorizations.c.project_id == project_id, action_authorizations.c.authorization_id == authorization_id, ) - .order_by( - action_authorizations.c.created_at.desc(), - literal_column("rowid", Integer()).desc(), - ) + .order_by(literal_column("rowid", Integer()).desc()) .limit(1) ) .mappings() @@ -320,9 +317,7 @@ def list_authorizations( statement = statement.where(action_authorizations.c.plan_id == plan_id) with read_connection(self._engine) as connection: rows = connection.execute( - statement.order_by( - action_authorizations.c.created_at, literal_column("rowid", Integer()) - ) + statement.order_by(literal_column("rowid", Integer())) ).mappings() values = tuple( AuthorizationEnvelopeRecord.model_validate(load(row["content_json"])) @@ -338,6 +333,22 @@ def list_authorizations( or self._resource_access.may_read_revision(project_id, latest[key].plan_revision_digest) ) + def has_authorization_successor( + self, project_id: str, authorization_id: str, revision_digest: str + ) -> bool: + with read_connection(self._engine) as connection: + rows = connection.execute( + select(action_authorizations.c.content_json).where( + action_authorizations.c.project_id == project_id, + action_authorizations.c.authorization_id == authorization_id, + ) + ).scalars() + return any( + AuthorizationEnvelopeRecord.model_validate(load(row)).supersedes_revision_digest + == revision_digest + for row in rows + ) + def append_audit(self, value: ActionAuditRecord) -> None: with write_connection(self._engine) as connection: connection.execute( @@ -374,9 +385,9 @@ def list_audit(self, project_id: str, subject_id: str | None) -> tuple[ActionAud ) if self._resource_access is None: return values - if self._ledger is None: - return () - heads = self._ledger.read_heads(project_id) + if self._ledger is None: + return () + heads = self._ledger.read_heads(project_id) return tuple( item for item in values diff --git a/src/thoth/application/commands/executions_full.py b/src/thoth/application/commands/executions_full.py index ff3cd0c..dc75bcc 100644 --- a/src/thoth/application/commands/executions_full.py +++ b/src/thoth/application/commands/executions_full.py @@ -30,6 +30,7 @@ from thoth.application.services.sandbox_service import SandboxService from thoth.domain.action_full import ActionPlanRecord from thoth.domain.auth import current_authenticated_actor +from thoth.domain.authorization_errors import AuthorizationAlreadyConsumed from thoth.domain.execution_full import PlanExecutionRecord, StepExecutionAttemptRecord from thoth.domain.sandbox import SandboxFailure, SandboxRunSpec from thoth.ports.action import ActionStorePort @@ -277,6 +278,10 @@ async def start(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: str(exc), data=data, ) from exc + except AuthorizationAlreadyConsumed as exc: + raise RpcApplicationError( + RpcErrorCode.DOMAIN_REJECTED, str(exc), data={"reason_code": exc.reason_code} + ) from exc except ValueError as exc: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, str(exc)) from exc return cast( @@ -316,6 +321,10 @@ async def resume(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: execution, attempts, commit = self._service.resume( current, self._plan(current), request.checkpoint_digest ) + except AuthorizationAlreadyConsumed as exc: + raise RpcApplicationError( + RpcErrorCode.DOMAIN_REJECTED, str(exc), data={"reason_code": exc.reason_code} + ) from exc except ValueError as exc: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, str(exc)) from exc return { @@ -367,6 +376,10 @@ async def retry(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: failed_attempt_id=request.failed_attempt_id, retry_reason=request.retry_reason, ) + except AuthorizationAlreadyConsumed as exc: + raise RpcApplicationError( + RpcErrorCode.DOMAIN_REJECTED, str(exc), data={"reason_code": exc.reason_code} + ) from exc except ValueError as exc: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, str(exc)) from exc return { @@ -391,6 +404,10 @@ async def reconcile(self, value: dict[str, JsonValue]) -> dict[str, JsonValue]: method=request.reconciliation_method, ) ) + except AuthorizationAlreadyConsumed as exc: + raise RpcApplicationError( + RpcErrorCode.DOMAIN_REJECTED, str(exc), data={"reason_code": exc.reason_code} + ) from exc except ValueError as exc: raise RpcApplicationError(RpcErrorCode.DOMAIN_REJECTED, str(exc)) from exc return { diff --git a/src/thoth/application/services/authorization_consumption.py b/src/thoth/application/services/authorization_consumption.py index 865d0a4..e1d8f76 100644 --- a/src/thoth/application/services/authorization_consumption.py +++ b/src/thoth/application/services/authorization_consumption.py @@ -7,6 +7,7 @@ from thoth.application.services.hypothesis_link_view import HypothesisLinkReader from thoth.domain.action_full import ActionAuditRecord, AuthorizationEnvelopeRecord from thoth.domain.auth import require_authenticated_authority +from thoth.domain.authorization_errors import AuthorizationAlreadyConsumed from thoth.domain.canonical import canonical_payload, domain_digest from thoth.ports.action import ActionStorePort from thoth.ports.governance import GovernanceStorePort @@ -28,7 +29,11 @@ def consume_authorization( with ledger.transaction(): current = store.read_authorization(expected.project_id, expected.authorization_id) if current is None or current.revision_digest != expected.revision_digest: - raise ValueError("authorization revision changed before consumption") + raise AuthorizationAlreadyConsumed() + if store.has_authorization_successor( + expected.project_id, expected.authorization_id, expected.revision_digest + ): + raise AuthorizationAlreadyConsumed() if current.state != "APPROVED": raise ValueError("authorization is not approved") if current.exact_scope_digest != exact_scope_digest: diff --git a/src/thoth/application/services/authorization_currentness.py b/src/thoth/application/services/authorization_currentness.py index 89ec585..ca5b7ba 100644 --- a/src/thoth/application/services/authorization_currentness.py +++ b/src/thoth/application/services/authorization_currentness.py @@ -82,10 +82,20 @@ def approved_authorization( ) -> AuthorizationEnvelopeRecord | None: """The approved envelope that still covers this step, if any.""" + by_id: dict[str, dict[str, AuthorizationEnvelopeRecord]] = {} + for item in envelopes: + by_id.setdefault(item.authorization_id, {})[item.revision_digest] = item + current: list[AuthorizationEnvelopeRecord] = [] + for revisions in by_id.values(): + superseded = {item.supersedes_revision_digest for item in revisions.values()} + tips = [item for digest, item in revisions.items() if digest not in superseded] + # Ambiguous branches must not revive an earlier APPROVED revision. + if len(tips) == 1: + current.append(tips[0]) return next( ( item - for item in envelopes + for item in current if item.step_id == step.get("step_id") and item.state == "APPROVED" and authorization_applies(item, plan, step) diff --git a/src/thoth/domain/authorization_errors.py b/src/thoth/domain/authorization_errors.py new file mode 100644 index 0000000..38c2beb --- /dev/null +++ b/src/thoth/domain/authorization_errors.py @@ -0,0 +1,13 @@ +"""Typed rejection of an approval revision that cannot be consumed again.""" + +from thoth.domain.errors import DomainError + + +class AuthorizationAlreadyConsumed(DomainError): + reason_code = "AUTHORIZATION_ALREADY_CONSUMED" + + def __init__(self) -> None: + super().__init__( + "이미 사용되었거나 최신 상태가 아닌 승인입니다. 승인을 다시 확인하세요. / " + "This approval was already used or is no longer current. Check the approval again." + ) diff --git a/src/thoth/ports/action.py b/src/thoth/ports/action.py index 0ec3e2b..3376e41 100644 --- a/src/thoth/ports/action.py +++ b/src/thoth/ports/action.py @@ -34,6 +34,9 @@ def read_authorization( def list_authorizations( self, project_id: str, plan_id: str | None ) -> tuple[AuthorizationEnvelopeRecord, ...]: ... + def has_authorization_successor( + self, project_id: str, authorization_id: str, revision_digest: str + ) -> bool: ... def append_audit(self, value: ActionAuditRecord) -> None: ... def list_audit( self, project_id: str, subject_id: str | None diff --git a/tests/contract/test_async_http.py b/tests/contract/test_async_http.py index 3ce542e..37a663a 100644 --- a/tests/contract/test_async_http.py +++ b/tests/contract/test_async_http.py @@ -99,7 +99,9 @@ async def slow_handler(_value: dict[str, JsonValue]) -> dict[str, JsonValue]: application = create_app(bus) transport = httpx.ASGITransport(app=application) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: started = await client.post("/rpc/async", json=_request()) assert started.status_code == 200 payload = started.json() @@ -170,7 +172,9 @@ async def slow_handler(_value: dict[str, JsonValue]) -> dict[str, JsonValue]: application = create_app(bus, auth=auth) transport = httpx.ASGITransport(app=application) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: started = await client.post( "/rpc/async", json=_request(), diff --git a/tests/contract/test_desktop_web_mount.py b/tests/contract/test_desktop_web_mount.py index 0a38943..a98bdd1 100644 --- a/tests/contract/test_desktop_web_mount.py +++ b/tests/contract/test_desktop_web_mount.py @@ -12,7 +12,7 @@ def test_same_origin_serves_ui_and_healthz(tmp_path: Path, monkeypatch: object) (dist / "index.html").write_text("THOTH UI", encoding="utf-8") monkeypatch.setenv("THOTH_WEB_DIST", str(dist)) # type: ignore[union-attr] monkeypatch.setenv("THOTH_WORKSPACE", str(tmp_path / "ws")) # type: ignore[union-attr] - client = TestClient(create_app()) + client = TestClient(create_app(), base_url="http://127.0.0.1:8765") assert client.get("/healthz").json()["status"] == "ok" assert "THOTH UI" in client.get("/").text diff --git a/tests/contract/test_file_upload.py b/tests/contract/test_file_upload.py index 7b143c0..6ed749e 100644 --- a/tests/contract/test_file_upload.py +++ b/tests/contract/test_file_upload.py @@ -15,7 +15,7 @@ async def test_file_stage_writes_content_addressed_inbox_file( workspace = tmp_path / "workspace" monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) transport = httpx.ASGITransport(app=create_app()) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: response = await client.post( "/files/stage", files={"file": ("../계획서.md", b"# approved plan\n", "text/markdown")}, @@ -38,7 +38,7 @@ async def test_file_stage_bounds_name_and_reuses_verified_digest( monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) transport = httpx.ASGITransport(app=create_app()) name = f"{'a' * 300}.md" - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: first = await client.post( "/files/stage", files={"file": (name, b"bounded", "text/markdown")}, @@ -62,7 +62,7 @@ async def test_file_stage_rejects_unsupported_type( ) -> None: monkeypatch.setenv("THOTH_WORKSPACE", str(tmp_path / "workspace")) transport = httpx.ASGITransport(app=create_app()) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: response = await client.post( "/files/stage", files={"file": ("payload.exe", b"MZ", "application/octet-stream")}, @@ -79,7 +79,7 @@ async def test_file_stage_rejects_corrupt_existing_digest_path( workspace = tmp_path / "workspace" monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) transport = httpx.ASGITransport(app=create_app()) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: first = await client.post( "/files/stage", files={"file": ("plan.md", b"trusted", "text/markdown")}, @@ -105,7 +105,7 @@ async def test_file_stage_takes_yaml_as_plain_text_and_keeps_its_name( monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) transport = httpx.ASGITransport(app=create_app()) body = b"identity: example\nnumerator: 19\n" - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: response = await client.post( "/files/stage", files={"file": (name, body, "application/octet-stream")}, @@ -126,7 +126,7 @@ async def test_file_stage_still_rejects_other_types_next_to_yaml( ) -> None: monkeypatch.setenv("THOTH_WORKSPACE", str(tmp_path / "workspace")) transport = httpx.ASGITransport(app=create_app()) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: response = await client.post( "/files/stage", files={"file": (name, b"a: 1\n", "text/plain")}, diff --git a/tests/contract/test_http_async_route_contract.py b/tests/contract/test_http_async_route_contract.py index fed2c89..017770f 100644 --- a/tests/contract/test_http_async_route_contract.py +++ b/tests/contract/test_http_async_route_contract.py @@ -189,7 +189,9 @@ async def test_cancel_denies_each_owner_mismatch_without_touching_original_task( auth = StaticActorAuth({"owner": actor, "other": other}) async with ( running_app(tmp_path, auth) as (app, bundle, _blocker, _observed), - httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" + ) as client, ): started = await client.post( "/rpc/async", json=request_body(), headers={"authorization": "Bearer owner"} @@ -215,7 +217,9 @@ async def test_matching_owner_scope_order_can_cancel_and_get_requires_rpc(tmp_pa auth = StaticActorAuth({"owner": actor, "reordered": reordered}) async with ( running_app(tmp_path, auth) as (app, bundle, _blocker, _observed), - httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" + ) as client, ): started = await client.post( "/rpc/async", json=request_body(), headers={"authorization": "Bearer owner"} @@ -250,7 +254,9 @@ async def test_matching_owner_scope_order_can_cancel_and_get_requires_rpc(tmp_pa async def test_local_missing_or_wrong_project_read_and_empty_cancel(tmp_path: Path) -> None: async with ( running_app(tmp_path) as (app, _bundle, _blocker, _observed), - httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" + ) as client, ): started = await client.post("/rpc/async", json=request_body()) identifier = started.json()["result"]["operation_id"] @@ -300,7 +306,9 @@ def forbidden(*_args: object) -> CommandBus: async def test_local_identity_denial_never_claims_or_creates_a_task(tmp_path: Path) -> None: async with ( running_app(tmp_path) as (app, _bundle, _blocker, observed), - httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" + ) as client, ): response = await client.post("/rpc/async", json=request_body(actor_id="human:spoofed")) assert response.status_code == 200 @@ -314,7 +322,7 @@ async def test_lifespan_cancels_task_from_the_same_registry(tmp_path: Path) -> N async with ( app.router.lifespan_context(app), httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client, ): started = await client.post("/rpc/async", json=request_body()) @@ -342,7 +350,9 @@ def claim( async with ( running_app(tmp_path, local_actor_id="human:custom-local") as (app, _, blocker, observed), - httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" + ) as client, ): monkeypatch.setattr(http_app, "_claim_local_rpc", claim) response = await client.post("/rpc/async", json=request_body(actor_id="human:custom-local")) diff --git a/tests/contract/test_http_file_staging_contract.py b/tests/contract/test_http_file_staging_contract.py index 9a303fd..fbec36f 100644 --- a/tests/contract/test_http_file_staging_contract.py +++ b/tests/contract/test_http_file_staging_contract.py @@ -140,7 +140,7 @@ def workspace(_request: Request) -> Path: monkeypatch.setattr(http_app, "_workspace_path_for_request", workspace) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client: response = await client.post( "/files/stage", @@ -167,7 +167,7 @@ def forbidden(_request: Request) -> Path: monkeypatch.setattr(http_app, "_workspace_path_for_request", forbidden) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client: response = await client.post( "/files/stage", diff --git a/tests/contract/test_local_loopback_identity.py b/tests/contract/test_local_loopback_identity.py index 04d4059..fbdd0c6 100644 --- a/tests/contract/test_local_loopback_identity.py +++ b/tests/contract/test_local_loopback_identity.py @@ -48,7 +48,9 @@ async def handler(value: dict[str, JsonValue]) -> dict[str, JsonValue]: bus = CommandBus(registry, operations, SystemClock(), UuidIdGenerator()) transport = httpx.ASGITransport(app=create_app(bus)) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: denied = await client.post("/rpc", json=rpc("human:spoof")) payload = denied.json() @@ -81,7 +83,9 @@ async def handler(value: dict[str, JsonValue]) -> dict[str, JsonValue]: bus = CommandBus(registry, operations, SystemClock(), UuidIdGenerator()) transport = httpx.ASGITransport(app=create_app(bus)) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: accepted = await client.post("/rpc", json=rpc("human:local-user")) payload = accepted.json() diff --git a/tests/contract/test_local_request_guard.py b/tests/contract/test_local_request_guard.py new file mode 100644 index 0000000..9686124 --- /dev/null +++ b/tests/contract/test_local_request_guard.py @@ -0,0 +1,120 @@ +"""Local HTTP guard runs before parsing, runtime creation and file staging.""" + +from __future__ import annotations + +import importlib +import json +from pathlib import Path +from typing import cast + +import httpx +import pytest +from fastapi import Request, WebSocket +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect + +from thoth.adapters.http.app import create_app + + +@pytest.fixture +def local_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + workspace = tmp_path / "workspace" + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "LOCAL") + monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) + monkeypatch.setenv("THOTH_WEB_DIST", str(tmp_path / "missing-dist")) + return workspace + + +@pytest.mark.parametrize( + "method,path,headers,code", + ( + ("GET", "/healthz", {"host": "evil.example"}, "LOCAL_HOST_REJECTED"), + ("GET", "/", {"host": "evil.example"}, "LOCAL_HOST_REJECTED"), + ("POST", "/rpc", {"origin": "null"}, "LOCAL_ORIGIN_REJECTED"), + ("POST", "/rpc/query", {"content-type": "text/plain"}, "JSON_CONTENT_TYPE_REQUIRED"), + ("POST", "/rpc/async", {}, "JSON_CONTENT_TYPE_REQUIRED"), + ( + "POST", + "/operations/one/cancel", + {"origin": "https://localhost"}, + "LOCAL_ORIGIN_REJECTED", + ), + ( + "POST", + "/files/stage", + {"origin": "http://evil.example", "x-thoth-project-id": "p"}, + "LOCAL_ORIGIN_REJECTED", + ), + ("PUT", "/future-write", {"origin": "null"}, "LOCAL_ORIGIN_REJECTED"), + ("PATCH", "/future-write", {"origin": "null"}, "LOCAL_ORIGIN_REJECTED"), + ("DELETE", "/future-write", {"origin": "null"}, "LOCAL_ORIGIN_REJECTED"), + ), +) +def test_rejection_does_not_parse_or_open_workspace( + local_environment: Path, + monkeypatch: pytest.MonkeyPatch, + method: str, + path: str, + headers: dict[str, str], + code: str, +) -> None: + module = importlib.import_module("thoth.adapters.http.app") + + def forbidden(*args: object, **kwargs: object) -> None: + pytest.fail("request reached body/runtime/staging") + + monkeypatch.setattr(Request, "body", forbidden) + monkeypatch.setattr(module, "_resolve_bus_for_request", forbidden) + monkeypatch.setattr(module, "_stage_file", forbidden) + app = create_app() + with TestClient(app, base_url="http://localhost:8765") as client: + response = cast(httpx.Client, client).request(method, path, headers=headers, content=b"{") + assert response.status_code == (415 if code == "JSON_CONTENT_TYPE_REQUIRED" else 403) + assert response.json()["code"] == code + assert set(response.json()) == {"code", "message_ko", "message_en"} + assert "evil.example" not in response.text + assert not any(key.startswith("access-control-") for key in response.headers) + assert app.state.runtime is None and app.state.bus is None + assert app.state.background_tasks == {} + assert not local_environment.exists() + + +@pytest.mark.parametrize("headers", ({"host": "evil.example"}, {"origin": "null"})) +def test_future_websocket_is_rejected_before_handler( + local_environment: Path, headers: dict[str, str] +) -> None: + app = create_app() + + async def endpoint(socket: WebSocket) -> None: + pytest.fail("rejected websocket reached handler") + + app.add_api_websocket_route("/future-socket", endpoint) + with ( + TestClient(app, base_url="http://localhost:8765") as client, + pytest.raises(WebSocketDisconnect) as caught, + client.websocket_connect("ws://localhost:8765/future-socket", headers=headers), + ): + pytest.fail("rejected websocket accepted") + assert caught.value.code == 1008 + payload = json.loads(caught.value.reason) + assert payload["code"] == ( + "LOCAL_HOST_REJECTED" if "host" in headers else "LOCAL_ORIGIN_REJECTED" + ) + assert not local_environment.exists() + + +def test_future_websocket_without_origin_is_allowed(local_environment: Path) -> None: + app = create_app() + + async def endpoint(socket: WebSocket) -> None: + await socket.accept() + await socket.send_text("synthetic") + await socket.close() + + app.add_api_websocket_route("/future-socket", endpoint) + with ( + TestClient(app, base_url="http://localhost:8765") as client, + client.websocket_connect("ws://localhost:8765/future-socket") as socket, + ): + assert socket.receive_text() == "synthetic" + assert not local_environment.exists() diff --git a/tests/contract/test_transports.py b/tests/contract/test_transports.py index 8d561dc..38b5571 100644 --- a/tests/contract/test_transports.py +++ b/tests/contract/test_transports.py @@ -37,7 +37,7 @@ async def test_stdio_and_http_return_same_payload(command_bus: CommandBus) -> No ) transport = httpx.ASGITransport(app=create_app(command_bus)) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:8765") as client: http_response = await client.post( "/rpc", content=request, headers={"content-type": "application/json"} ) diff --git a/tests/integration/resource_scope_helpers.py b/tests/integration/resource_scope_helpers.py index e82b8b8..ae22905 100644 --- a/tests/integration/resource_scope_helpers.py +++ b/tests/integration/resource_scope_helpers.py @@ -150,7 +150,7 @@ async def scope_peer(original: ScopeHarness) -> AsyncGenerator[ScopeHarness]: ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: yield ScopeHarness( original.workspace, @@ -241,7 +241,7 @@ async def scope_harness( ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: tokens: dict[str, str] = {} for name, actor in actors.items(): diff --git a/tests/integration/test_a06_memory_storage_boundary.py b/tests/integration/test_a06_memory_storage_boundary.py index 1a664f0..0de8962 100644 --- a/tests/integration/test_a06_memory_storage_boundary.py +++ b/tests/integration/test_a06_memory_storage_boundary.py @@ -502,7 +502,7 @@ def expired_now(clock: SystemClock) -> datetime: monkeypatch.setattr(DeterministicRoleMemoryReviewer, "review", review) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: response = await client.post( "/rpc", diff --git a/tests/integration/test_a08_storage_authority.py b/tests/integration/test_a08_storage_authority.py index d028e86..4e3b759 100644 --- a/tests/integration/test_a08_storage_authority.py +++ b/tests/integration/test_a08_storage_authority.py @@ -287,9 +287,9 @@ async def test_finalization_fault_preserves_preparation_and_rolls_back_activatio (SandboxExecutionState.FAILED, "SEMANTIC finalize fault"), ) ), - max_retries=0, - improvement_evaluator=DeterministicIndependentImprovementEvaluator(), - ) + max_retries=0, + improvement_evaluator=DeterministicIndependentImprovementEvaluator(), + ) try: value( await runtime.bus.dispatch( @@ -513,9 +513,9 @@ async def test_repeated_successful_candidate_keeps_the_actual_active_registry( for _ in range(3) ) ), - max_retries=0, - improvement_evaluator=DeterministicIndependentImprovementEvaluator(), - ) + max_retries=0, + improvement_evaluator=DeterministicIndependentImprovementEvaluator(), + ) try: for ordinal in (1, 2): value( @@ -725,7 +725,7 @@ def expired_now(clock: SystemClock) -> datetime: ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: response = await client.post( "/rpc", diff --git a/tests/integration/test_a13_authenticated_multiplayer.py b/tests/integration/test_a13_authenticated_multiplayer.py index 5670ef9..082fb80 100644 --- a/tests/integration/test_a13_authenticated_multiplayer.py +++ b/tests/integration/test_a13_authenticated_multiplayer.py @@ -71,7 +71,9 @@ async def test_http_rpc_denies_unauthenticated_request_before_bus_io(tmp_path: P ) app = create_app(runtime.bus, auth=DenyAllAuth()) transport = httpx.ASGITransport(app=app) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: response = await client.post( "/rpc", json={ @@ -166,7 +168,9 @@ async def test_two_authenticated_scoped_actors_branch_merge_and_bind_revision_re ids=UuidIdGenerator(), ) transport = httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: sessions: dict[str, dict[str, JsonValue]] = {} for actor in (alice, bob, scoped): response = await client.post( diff --git a/tests/integration/test_a13_operation_scope.py b/tests/integration/test_a13_operation_scope.py index 52830ea..a2bcfb4 100644 --- a/tests/integration/test_a13_operation_scope.py +++ b/tests/integration/test_a13_operation_scope.py @@ -131,7 +131,9 @@ async def harness(tmp_path: Path, request: pytest.FixtureRequest) -> AsyncIterat assert legacy.result is not None transport = httpx.ASGITransport(app=create_app(runtime.bus, auth=auth_service(runtime))) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: for alias in (*scopes, "renewed"): actor = "alpha" if alias == "renewed" else alias issued = await client.post( @@ -348,7 +350,9 @@ async def test_operation_scope_and_result_persist_after_reopen( reopened = create_runtime(tmp_path / "scope") transport = httpx.ASGITransport(app=create_app(reopened.bus, auth=auth_service(reopened))) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: restored = ScopeHarness( reopened, client, diff --git a/tests/integration/test_a13_security_boundaries.py b/tests/integration/test_a13_security_boundaries.py index 2af9bc9..7e98250 100644 --- a/tests/integration/test_a13_security_boundaries.py +++ b/tests/integration/test_a13_security_boundaries.py @@ -93,7 +93,9 @@ async def test_local_loopback_rejects_protected_actor_ref_override(tmp_path: Pat application = create_app(runtime.bus) transport = httpx.ASGITransport(app=application) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: response = await client.post( "/rpc", json={ @@ -201,7 +203,9 @@ async def test_authenticated_upload_path_is_project_scoped_at_stage_and_connect( auth=StaticProjectAuth({"token-a": context_a, "token-b": context_b}), ) transport = httpx.ASGITransport(app=application) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: staged = await client.post( "/files/stage", files={"file": ("plan.md", b"# project A\n", "text/markdown")}, diff --git a/tests/integration/test_approval_clock_rollback.py b/tests/integration/test_approval_clock_rollback.py new file mode 100644 index 0000000..39ea5d4 --- /dev/null +++ b/tests/integration/test_approval_clock_rollback.py @@ -0,0 +1,273 @@ +"""F-03 regression: one approval stays consumed even when the clock moves backwards. + +Only SystemClock is changed, never the system clock. Existing R3 helpers use the +local-metadata-only profile; model resolution, network sends and processes are forbidden. +""" + +from __future__ import annotations + +import json +import subprocess +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import NoReturn, cast + +import httpx +import pytest +from sqlalchemy import event, func, literal_column, select +from sqlalchemy.engine import ExceptionContext +from tests.integration import storage_coverage_helpers +from tests.integration.test_a04_storage_authority import prepare_r3, start_request + +from thoth.adapters.runtime import SystemClock +from thoth.adapters.storage.action import SqliteActionStore +from thoth.adapters.storage.schema import ( + action_authorizations, + execution_effects, + step_execution_attempts, +) +from thoth.apps.runtime import AppRuntime +from thoth.domain.action_full import AuthorizationEnvelopeRecord +from thoth.domain.execution_full import StepExecutionAttemptRecord +from thoth.ports.model import ModelPort + +APPROVED_AT = datetime(2026, 10, 19, 12, tzinfo=UTC) + + +class NoModels: + calls = 0 + + def resolve(self, *, provider: str, model: str | None) -> ModelPort: + del provider, model + self.calls += 1 + raise AssertionError("model resolution is forbidden in the clock baseline") + + +@pytest.fixture +def isolated_metadata_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[NoModels]: + models = NoModels() + runtimes: list[AppRuntime] = [] + original = storage_coverage_helpers.create_runtime + forbidden_calls: list[str] = [] + + def runtime(workspace: Path) -> AppRuntime: + # Explicit resolver avoids automatic account-catalog refresh during setup. + result = original(workspace, model_resolver=models) + runtimes.append(result) + return result + + def forbidden_process(*args: object, **kwargs: object) -> NoReturn: + forbidden_calls.append("process") + raise AssertionError("process creation is forbidden in the clock baseline") + + def forbidden_network(*args: object, **kwargs: object) -> NoReturn: + forbidden_calls.append("network") + raise AssertionError("network send is forbidden in the clock baseline") + + for name in ("HOME", "USERPROFILE", "LOCALAPPDATA", "APPDATA", "CODEX_HOME"): + monkeypatch.setenv(name, str(tmp_path / "synthetic-home" / name.lower())) + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "LOCAL") + monkeypatch.setattr(storage_coverage_helpers, "create_runtime", runtime) + monkeypatch.setattr(subprocess, "Popen", forbidden_process) + monkeypatch.setattr(httpx.Client, "send", forbidden_network) + monkeypatch.setattr(httpx.AsyncClient, "send", forbidden_network) + try: + yield models + finally: + for created in reversed(runtimes): + created.close() + assert models.calls == 0 + assert forbidden_calls == [] + + +@dataclass +class InjectedClock: + current: datetime = APPROVED_AT + step: timedelta = timedelta(0) + + def now(self) -> datetime: + result = self.current + self.current += self.step + return result + + +def history( + runtime: AppRuntime, project: str, authorization: str +) -> list[AuthorizationEnvelopeRecord]: + with runtime.ledger.engine.connect() as connection: + rows = connection.execute( + select(action_authorizations.c.content_json) + .where( + action_authorizations.c.project_id == project, + action_authorizations.c.authorization_id == authorization, + ) + .order_by(literal_column("rowid")) + ).scalars() + return [AuthorizationEnvelopeRecord.model_validate_json(str(row)) for row in rows] + + +def attempts(runtime: AppRuntime, project: str) -> list[StepExecutionAttemptRecord]: + with runtime.ledger.engine.connect() as connection: + rows = connection.execute( + select(step_execution_attempts.c.content_json).where( + step_execution_attempts.c.project_id == project + ) + ).scalars() + return [StepExecutionAttemptRecord.model_validate_json(str(row)) for row in rows] + + +def diagnostics(workspace: Path) -> dict[str, object]: + path = workspace / "diagnostics" / "internal-failures.jsonl" + types: list[str] = [] + if path.exists(): + for line in path.read_text(encoding="utf-8").splitlines(): + record: object = json.loads(line) + if isinstance(record, dict): + kind = cast(dict[str, object], record).get("exception_type") + if isinstance(kind, str): + types.append(kind) + return {"exists": path.exists(), "exception_types": types} + + +# The normal response contract also applies to both rollback clocks. +@pytest.mark.parametrize("scenario", ("normal", "rollback-fixed", "rollback-advancing")) +async def test_current_approval_clock_behavior( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + isolated_metadata_only: NoModels, + scenario: str, +) -> None: + clock = InjectedClock() + + def injected_now(_clock: SystemClock) -> datetime: + return clock.now() + + monkeypatch.setattr(SystemClock, "now", injected_now) + workspace = tmp_path / "workspace" + runtime, project, plan, approval = await prepare_r3(workspace) + store = SqliteActionStore(runtime.ledger.engine) + original = store.read_authorization(project, str(approval["authorization_id"])) + assert original is not None and original.state == "APPROVED" and original.single_use + assert original.created_at == APPROVED_AT + assert not [ + row + for row in history(runtime, project, original.authorization_id) + if row.state == "CONSUMED" + ] + sql_errors: list[dict[str, object]] = [] + + def observe_error(context: ExceptionContext) -> None: + wrapper = context.sqlalchemy_exception + original_error = context.original_exception + # Do not retain SQL parameters, raw exception text, paths or provider output. + sql_errors.append( + { + "sqlalchemy_type": None if wrapper is None else type(wrapper).__name__, + "dbapi_type": type(original_error).__name__, + "action_audit_insert": (context.statement or "").startswith( + "INSERT INTO action_audit" + ), + "audit_digest_unique": str(original_error) + == "UNIQUE constraint failed: action_audit.event_digest", + } + ) + + event.listen(runtime.ledger.engine, "handle_error", observe_error) + try: + clock.current = ( + APPROVED_AT + timedelta(seconds=1) + if scenario == "normal" + else APPROVED_AT - timedelta(minutes=1) + ) + clock.step = timedelta(0) if scenario == "rollback-fixed" else timedelta(milliseconds=1) + first = await runtime.bus.dispatch(start_request(runtime, project, plan, "clock-first")) + assert first.error is None and first.result is not None + current_first = store.read_authorization(project, original.authorization_id) + assert current_first is not None + first_consumed = [ + row + for row in history(runtime, project, original.authorization_id) + if row.state == "CONSUMED" + ] + assert len(first_consumed) == 1 + assert len(attempts(runtime, project)) == 1 + second = await runtime.bus.dispatch(start_request(runtime, project, plan, "clock-second")) + current_second = store.read_authorization(project, original.authorization_id) + assert current_second is not None + rows = history(runtime, project, original.authorization_id) + consumed = [row for row in rows if row.state == "CONSUMED"] + persisted_attempts = attempts(runtime, project) + listed = store.list_authorizations(project, str(plan["plan_id"])) + with runtime.ledger.engine.connect() as connection: + effect_rows = int( + connection.execute( + select(func.count()) + .select_from(execution_effects) + .where(execution_effects.c.project_id == project) + ).scalar_one() + ) + report: dict[str, object] = { + "scenario": scenario, + "first_code": None if first.error is None else int(first.error.code), + "second_code": None if second.error is None else int(second.error.code), + "current_after_first": current_first.state, + "current_after_second": current_second.state, + "current_still_original_approval": current_second.revision_digest + == original.revision_digest, + "listed_states": [row.state for row in listed], + "stored_states": [row.state for row in rows], + "consumed_rows": len(consumed), + "attempt_rows": len(persisted_attempts), + "effect_rows": effect_rows, + "consumed_times": [row.created_at.isoformat() for row in consumed], + "all_consumed_from_original": all( + row.supersedes_revision_digest == original.revision_digest for row in consumed + ), + "diagnostics": diagnostics(workspace), + "sqlalchemy_observation": sql_errors, + "model_calls": isolated_metadata_only.calls, + "profile": "local-metadata-only", + } + print("VF03 " + json.dumps(report, sort_keys=True)) + assert all( + row.single_use and row.exact_scope_digest == original.exact_scope_digest + for row in consumed + ) + # The existing attempt contract binds exact scope, not an authorization revision. + assert {row.authorization_digest for row in persisted_attempts} == { + original.exact_scope_digest + } + assert all(row.effect_state == "POSSIBLE" for row in persisted_attempts) + assert effect_rows == 0 + if scenario == "normal": + assert current_first.state == current_second.state == "CONSUMED" + assert len(consumed) == len(persisted_attempts) == 1 + assert second.error is None and second.result is not None + second_value = second.result["value"] + assert isinstance(second_value, dict) + execution = second_value["execution"] + assert isinstance(execution, dict) + assert execution["protected_steps"] == ["step:r3"] + assert execution["executable_steps"] == [] and execution["attempt_refs"] == [] + assert sql_errors == [] + else: + assert current_first.state == current_second.state == "CONSUMED" + assert current_second.revision_digest == consumed[0].revision_digest + assert current_second.revision_digest != original.revision_digest + assert [row.state for row in listed] == ["CONSUMED"] + assert clock.current < APPROVED_AT + assert len(consumed) == len(persisted_attempts) == 1 + assert consumed[0].created_at < APPROVED_AT + assert consumed[0].supersedes_revision_digest == original.revision_digest + assert second.error is None and second.result is not None + second_value = second.result["value"] + assert isinstance(second_value, dict) + execution = second_value["execution"] + assert isinstance(execution, dict) + assert execution["protected_steps"] == ["step:r3"] + assert execution["executable_steps"] == [] and execution["attempt_refs"] == [] + assert sql_errors == [] + finally: + event.remove(runtime.ledger.engine, "handle_error", observe_error) diff --git a/tests/integration/test_authorization_consumption_guards.py b/tests/integration/test_authorization_consumption_guards.py new file mode 100644 index 0000000..fd2b5d1 --- /dev/null +++ b/tests/integration/test_authorization_consumption_guards.py @@ -0,0 +1,181 @@ +"""Stale and concurrent consumers receive a typed rejection without another write.""" + +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta +from pathlib import Path +from threading import Barrier, Lock +from typing import Any, cast + +import pytest +from tests.integration import storage_coverage_helpers +from tests.integration.storage_coverage_helpers import domain_snapshot +from tests.integration.test_a04_storage_authority import prepare_r3, start_request +from tests.integration.test_approval_clock_rollback import ( + APPROVED_AT, + InjectedClock, + NoModels, + history, +) +from tests.integration.test_approval_clock_rollback import ( + isolated_metadata_only as isolated_metadata_only, +) + +from thoth.adapters.runtime import SystemClock +from thoth.adapters.storage.action import SqliteActionStore +from thoth.application.services.action_service import ActionService +from thoth.apps.runtime import AppRuntime +from thoth.domain.action_full import AuthorizationEnvelopeRecord +from thoth.domain.authorization_errors import AuthorizationAlreadyConsumed +from thoth.protocol.jsonrpc import RpcErrorCode + + +@pytest.fixture(autouse=True) +def fixed_clock(monkeypatch: pytest.MonkeyPatch) -> None: + def now(_clock: SystemClock) -> datetime: + return APPROVED_AT + + monkeypatch.setattr(SystemClock, "now", now) + + +def action_service(runtime: AppRuntime) -> ActionService: + # Test-only access to the actual service wired to the ordinary RPC handler. + handler = cast(Any, runtime.bus)._registry.resolve("execution/start") + return cast(ActionService, handler.__self__._service._action_service) + + +@pytest.mark.parametrize("stale_current_read", (False, True)) +async def test_stale_expected_and_successor_guard_leave_no_new_domain_write( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + isolated_metadata_only: NoModels, + stale_current_read: bool, +) -> None: + runtime, project, plan, grant = await prepare_r3(tmp_path / "workspace") + store = SqliteActionStore(runtime.ledger.engine) + expected = store.read_authorization(project, str(grant["authorization_id"])) + assert expected is not None + first = await runtime.bus.dispatch(start_request(runtime, project, plan, "first")) + assert first.error is None + before = domain_snapshot(runtime.ledger.engine) + checked_successor: list[bool] = [] + read = SqliteActionStore.read_authorization + has_successor = SqliteActionStore.has_authorization_successor + + def stale_read( + self: SqliteActionStore, project_id: str, authorization_id: str + ) -> AuthorizationEnvelopeRecord | None: + if ( + stale_current_read + and project_id == project + and authorization_id == expected.authorization_id + ): + return expected + return read(self, project_id, authorization_id) + + def observe_successor( + self: SqliteActionStore, project_id: str, authorization_id: str, digest: str + ) -> bool: + found = has_successor(self, project_id, authorization_id, digest) + checked_successor.append(found) + return found + + monkeypatch.setattr(SqliteActionStore, "read_authorization", stale_read) + monkeypatch.setattr(SqliteActionStore, "has_authorization_successor", observe_successor) + with pytest.raises(AuthorizationAlreadyConsumed) as caught: + action_service(runtime).consume_authorization( + expected, exact_scope_digest=expected.exact_scope_digest + ) + assert caught.value.reason_code == "AUTHORIZATION_ALREADY_CONSUMED" + assert checked_successor == ([True] if stale_current_read else []) + assert domain_snapshot(runtime.ledger.engine) == before + assert isolated_metadata_only.calls == 0 + + +async def test_stale_selection_is_a_typed_rpc_rejection_not_an_audit_collision( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + isolated_metadata_only: NoModels, +) -> None: + runtime, project, plan, grant = await prepare_r3(tmp_path / "workspace") + expected = SqliteActionStore(runtime.ledger.engine).read_authorization( + project, str(grant["authorization_id"]) + ) + assert expected is not None + first = await runtime.bus.dispatch(start_request(runtime, project, plan, "first")) + assert first.error is None + before = domain_snapshot(runtime.ledger.engine) + original = SqliteActionStore.list_authorizations + + def stale_selection( + self: SqliteActionStore, project_id: str, plan_id: str | None + ) -> tuple[AuthorizationEnvelopeRecord, ...]: + if project_id == project and plan_id == plan["plan_id"]: + return (expected,) + return original(self, project_id, plan_id) + + monkeypatch.setattr(SqliteActionStore, "list_authorizations", stale_selection) + response = await runtime.bus.dispatch(start_request(runtime, project, plan, "stale")) + assert response.error is not None and response.error.code == RpcErrorCode.DOMAIN_REJECTED + assert response.error.data == {"reason_code": "AUTHORIZATION_ALREADY_CONSUMED"} + assert domain_snapshot(runtime.ledger.engine) == before + assert ( + len( + [ + row + for row in history(runtime, project, expected.authorization_id) + if row.state == "CONSUMED" + ] + ) + == 1 + ) + assert isolated_metadata_only.calls == 0 + + +async def test_two_real_consumers_have_one_winner_and_one_typed_rejection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + isolated_metadata_only: NoModels, +) -> None: + workspace = tmp_path / "workspace" + first, project, _plan, grant = await prepare_r3(workspace) + second = storage_coverage_helpers.create_runtime(workspace) + expected = SqliteActionStore(first.ledger.engine).read_authorization( + project, str(grant["authorization_id"]) + ) + assert expected is not None + barrier = Barrier(2) + clock_lock = Lock() + clock = InjectedClock(APPROVED_AT - timedelta(minutes=1), timedelta(milliseconds=1)) + + def advancing_rollback(_clock: SystemClock) -> datetime: + with clock_lock: + return clock.now() + + monkeypatch.setattr(SystemClock, "now", advancing_rollback) + + def consume(runtime: AppRuntime) -> tuple[str, dict[str, tuple[str, ...]]]: + barrier.wait(timeout=10) + try: + result = action_service(runtime).consume_authorization( + expected, exact_scope_digest=expected.exact_scope_digest + ) + assert result.state == "CONSUMED" + state = "CONSUMED" + except AuthorizationAlreadyConsumed as exc: + state = exc.reason_code + return state, domain_snapshot(runtime.ledger.engine) + + with ThreadPoolExecutor(max_workers=2) as pool: + jobs = [pool.submit(consume, runtime) for runtime in (first, second)] + outcomes = [job.result(timeout=20) for job in jobs] + assert sorted(state for state, _snapshot in outcomes) == [ + "AUTHORIZATION_ALREADY_CONSUMED", + "CONSUMED", + ] + assert outcomes[0][1] == outcomes[1][1] == domain_snapshot(first.ledger.engine) + consumed = [ + row for row in history(first, project, expected.authorization_id) if row.state == "CONSUMED" + ] + assert len(consumed) == 1 and consumed[0].supersedes_revision_digest == expected.revision_digest + assert consumed[0].created_at < APPROVED_AT and clock.current < APPROVED_AT + assert isolated_metadata_only.calls == 0 diff --git a/tests/integration/test_codex_auth_nonblocking.py b/tests/integration/test_codex_auth_nonblocking.py index 6349078..ca68e72 100644 --- a/tests/integration/test_codex_auth_nonblocking.py +++ b/tests/integration/test_codex_auth_nonblocking.py @@ -31,16 +31,16 @@ def slow(_self: LocalModelCredentials) -> list[dict[str, object]]: try: created = await runtime.bus.dispatch( - request( - "project/create", - "create", - { - "project_id": "project:nonblocking", - "name": "Saved", - "cutoff_at": "2026-09-01T00:00:00Z", - }, - ) + request( + "project/create", + "create", + { + "project_id": "project:nonblocking", + "name": "Saved", + "cutoff_at": "2026-09-01T00:00:00Z", + }, ) + ) value(created) assert created.result is not None operation_id = created.result["operation_id"] @@ -56,7 +56,8 @@ def slow(_self: LocalModelCredentials) -> list[dict[str, object]]: ) assert await asyncio.to_thread(entered.wait, 3) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=create_app(runtime.bus)), base_url="http://test" + transport=httpx.ASGITransport(app=create_app(runtime.bus)), + base_url="http://127.0.0.1:8765", ) as client: health = await asyncio.wait_for(client.get("/healthz"), timeout=1) saved = await asyncio.wait_for( @@ -127,9 +128,7 @@ def no_db_transaction(method: str, params: dict[str, object]) -> dict[str, objec assert openai["execution_verified"] is False settings = value( await runtime.bus.dispatch( - request( - "model/settings/read", "settings", {"project_id": "project:codex-rpc"} - ) + request("model/settings/read", "settings", {"project_id": "project:codex-rpc"}) ) ) assert any(item["model"] == "synthetic-codex" for item in settings["model_options"]) diff --git a/tests/integration/test_counter_cleanup_cancel_preserves_prior.py b/tests/integration/test_counter_cleanup_cancel_preserves_prior.py index 3f3f4c2..1cd7f40 100644 --- a/tests/integration/test_counter_cleanup_cancel_preserves_prior.py +++ b/tests/integration/test_counter_cleanup_cancel_preserves_prior.py @@ -125,7 +125,7 @@ def state() -> tuple[list[tuple[str, str]], list[str]]: ) app = create_app(runtime.bus) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client: response = await client.post( "/rpc/async", diff --git a/tests/integration/test_decision_object_restore_fail_closed.py b/tests/integration/test_decision_object_restore_fail_closed.py index 11ef15c..b717d51 100644 --- a/tests/integration/test_decision_object_restore_fail_closed.py +++ b/tests/integration/test_decision_object_restore_fail_closed.py @@ -196,7 +196,7 @@ async def test_read_only_actor_cannot_restore_typed_decision(tmp_path: Path) -> ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: issued = await client.post( "/auth/session", diff --git a/tests/integration/test_local_http_origin_baseline.py b/tests/integration/test_local_http_origin_baseline.py new file mode 100644 index 0000000..c4a8bd4 --- /dev/null +++ b/tests/integration/test_local_http_origin_baseline.py @@ -0,0 +1,347 @@ +"""V-F04b: preserve local HTTP behavior and reject hostile headers before writes. + +The original 57 observations now pin the approved guard behavior. TestClient does +not prove browser delivery, DNS rebinding, preflight behavior or response access. +Only temporary workspaces and real local command handlers are used. Each MATRIX +line records committed row-count changes (not SQL statement counts). +""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import cast + +import httpx +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from thoth.adapters.http.app import create_app +from thoth.apps.runtime import AppRuntime, create_runtime +from thoth.protocol.bus import DispatchTicket +from thoth.protocol.jsonrpc import JsonRpcRequest, JsonRpcResponse + + +@dataclass(frozen=True) +class RequestShape: + name: str + host: str = "127.0.0.1:8765" + origin: str | None = None + content_type: str = "application/json" + + def headers(self) -> dict[str, str]: + result = {"host": self.host, "content-type": self.content_type} + if self.origin is not None: + result["origin"] = self.origin + return result + + +SHAPES = ( + RequestShape("ui-ipv4", origin="http://127.0.0.1:8765"), + RequestShape("ui-localhost", "localhost:8765", "http://localhost:8765"), + RequestShape("ui-ipv6", "[::1]:8765", "http://[::1]:8765"), + RequestShape("vite-ipv4", "127.0.0.1:5173", "http://127.0.0.1:5173"), + RequestShape("vite-localhost", "localhost:5173", "http://localhost:5173"), + RequestShape("local-no-origin"), + RequestShape("plain-no-origin", content_type="text/plain"), + RequestShape("foreign-origin-json", origin="http://evil.example"), + RequestShape("foreign-origin-plain", origin="http://evil.example", content_type="text/plain"), + RequestShape("foreign-host-only", "evil.example:8765"), + RequestShape("rebind-host-origin", "evil.example:8765", "http://evil.example:8765"), +) + + +@pytest.fixture +def local_api( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> Iterator[tuple[AppRuntime, TestClient, Path]]: + workspace = tmp_path / "workspace" + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "LOCAL") + monkeypatch.setenv("THOTH_WORKSPACE", str(workspace)) + monkeypatch.setenv("THOTH_WEB_DIST", str(tmp_path / "missing-dist")) + # Keep optional provider discovery away from the developer's credentials. + for name in ("HOME", "USERPROFILE", "LOCALAPPDATA", "APPDATA", "CODEX_HOME"): + monkeypatch.setenv(name, str(tmp_path / "synthetic-home" / name.lower())) + runtime = create_runtime(workspace) + try: + with TestClient(create_app(runtime.bus), base_url="http://127.0.0.1:8765") as client: + yield runtime, client, workspace + finally: + runtime.close() + + +def _request(method: str, key: str, project_id: str) -> dict[str, object]: + value: dict[str, object] = {"project_id": project_id} + if method == "project/create": + value.update(name="Synthetic HTTP baseline", cutoff_at="2026-10-19T00:00:00Z") + return { + "jsonrpc": "2.0", + "id": key, + "method": method, + "params": {"_meta": {"idempotencyKey": key}, "input": value}, + } + + +def _counts(runtime: AppRuntime) -> dict[str, int]: + with runtime.ledger.engine.connect() as connection: + tables = connection.exec_driver_sql( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'" + ).scalars() + return { + name: int( + connection.exec_driver_sql( + 'SELECT count(*) FROM "' + name.replace('"', '""') + '"' + ).scalar_one() + ) + for name in tables + } + + +def _record( + shape: RequestShape, + endpoint: str, + method: str, + response: httpx.Response, + before: dict[str, int], + after: dict[str, int], + **details: object, +) -> dict[str, int]: + delta = {name: after[name] - count for name, count in before.items() if after[name] != count} + print( + "MATRIX " + + json.dumps( + { + "shape": shape.name, + "endpoint": endpoint, + "method": method, + "headers": shape.headers(), + "http_status": response.status_code, + "cors_headers": { + k: v for k, v in response.headers.items() if k.startswith("access-control-") + }, + "row_deltas": delta, + "net_new_rows": sum(delta.values()), + **details, + }, + sort_keys=True, + ) + ) + return delta + + +@pytest.mark.parametrize("shape", SHAPES, ids=lambda shape: shape.name) +def test_local_http_header_baseline( + local_api: tuple[AppRuntime, TestClient, Path], shape: RequestShape +) -> None: + runtime, client, _workspace = local_api + http = cast(httpx.Client, client) + if shape.name in { + "plain-no-origin", + "foreign-origin-json", + "foreign-origin-plain", + "foreign-host-only", + "rebind-host-origin", + }: + _denied_matrix(runtime, http, shape) + return + assert client.portal is not None + for endpoint in ("/rpc", "/rpc/async"): + project_id = "project:" + shape.name + ("-async" if endpoint.endswith("async") else "-sync") + before = _counts(runtime) + response = http.post( + endpoint, + content=json.dumps(_request("project/create", project_id, project_id)), + headers=shape.headers(), + ) + assert response.status_code == 200 + assert response.json().get("error") is None, response.text + operation_id = response.json()["result"]["operation_id"] + + async def finish() -> None: + application = cast(FastAPI, client.app) + background_tasks = cast( + dict[str, asyncio.Task[JsonRpcResponse]], application.state.background_tasks + ) + tasks = tuple(background_tasks.values()) + await asyncio.gather(*tasks) + + client.portal.call(finish) + operation = runtime.bus.read_operation(operation_id) + assert operation is not None and operation.state.value == "SUCCEEDED" + after = _counts(runtime) + delta = _record(shape, endpoint, "project/create", response, before, after, executed=True) + assert delta["projects"] == 1 and delta["operations"] == 1 + # Independent ordinary HTTP readback confirms the persisted domain effect. + readback = http.post("/rpc/query", json=_request("project/read", "readback", project_id)) + assert readback.json()["result"]["value"]["project_id"] == project_id + + before = _counts(runtime) + read = http.post( + "/rpc/query", + content=json.dumps(_request("project/read", "read-probe", project_id)), + headers=shape.headers(), + ) + assert read.status_code == 200 + assert read.json()["result"]["value"]["project_id"] == project_id + assert not _record( + shape, "/rpc/query", "project/read", read, before, _counts(runtime), executed=True + ) + + absent_id = "project:query-must-not-write" + rejected = http.post( + "/rpc/query", + content=json.dumps(_request("project/create", "query-write", absent_id)), + headers=shape.headers(), + ) + assert rejected.status_code == 200 + assert rejected.json()["error"]["code"] == -32601 + assert rejected.json()["error"]["message"] == "method is not a read query" + assert not _record( + shape, + "/rpc/query", + "project/create", + rejected, + before, + _counts(runtime), + executed=False, + rejection="method is not a read query", + ) + + # Real claim + real cancellation handler, without starting any model or worker. + ticket = runtime.bus.claim( + JsonRpcRequest.model_validate(_request("project/create", "cancel-target", project_id)) + ) + assert isinstance(ticket, DispatchTicket) + target_id = ticket.operation.operation_id + assert ticket.operation.state.value == "RUNNING" + before = _counts(runtime) + cancel = http.post( + f"/operations/{target_id}/cancel", + content=json.dumps({"project_id": project_id}), + headers=shape.headers(), + ) + assert cancel.status_code == 200 + assert cancel.json()["result"]["value"]["cancelled"] is True + target = runtime.bus.read_operation(target_id) + assert target is not None and target.state.value == "CANCELLED" + delta = _record( + shape, + "/operations//cancel", + "operation/cancel", + cancel, + before, + _counts(runtime), + executed=True, + updated_target_rows=1, + target_before="RUNNING", + target_after="CANCELLED", + ) + assert delta["operations"] == 1 and "projects" not in delta + + +def _assert_denied(response: httpx.Response, shape: RequestShape) -> None: + expected = { + "plain-no-origin": (415, "JSON_CONTENT_TYPE_REQUIRED"), + "foreign-origin-json": (403, "LOCAL_ORIGIN_REJECTED"), + "foreign-origin-plain": (403, "LOCAL_ORIGIN_REJECTED"), + "foreign-host-only": (403, "LOCAL_HOST_REJECTED"), + "rebind-host-origin": (403, "LOCAL_HOST_REJECTED"), + }[shape.name] + assert (response.status_code, response.json()["code"]) == expected + assert response.json()["message_ko"] and response.json()["message_en"] + assert "evil.example" not in response.text + assert not any(key.startswith("access-control-") for key in response.headers) + + +def _denied_matrix(runtime: AppRuntime, http: httpx.Client, shape: RequestShape) -> None: + project_id = "project:denied" + for endpoint, method in ( + ("/rpc", "project/create"), + ("/rpc/async", "project/create"), + ("/rpc/query", "project/read"), + ("/rpc/query", "project/create"), + ): + before = _counts(runtime) + response = http.post( + endpoint, + content=json.dumps(_request(method, endpoint + method, project_id)), + headers=shape.headers(), + ) + _assert_denied(response, shape) + assert not _record( + shape, + endpoint, + method, + response, + before, + _counts(runtime), + executed=False, + rejection=response.json()["code"], + ) + assert _counts(runtime)["projects"] == 0 + ticket = runtime.bus.claim( + JsonRpcRequest.model_validate(_request("project/create", "cancel-target", project_id)) + ) + assert isinstance(ticket, DispatchTicket) + before = _counts(runtime) + original = runtime.bus.read_operation(ticket.operation.operation_id) + response = http.post( + f"/operations/{ticket.operation.operation_id}/cancel", + content=json.dumps({"project_id": project_id}), + headers=shape.headers(), + ) + _assert_denied(response, shape) + assert runtime.bus.read_operation(ticket.operation.operation_id) == original + assert not _record( + shape, + "/operations//cancel", + "operation/cancel", + response, + before, + _counts(runtime), + executed=False, + rejection=response.json()["code"], + updated_target_rows=0, + target_before="RUNNING", + target_after="RUNNING", + ) + + +@pytest.mark.parametrize("with_project_header", (False, True)) +def test_file_staging_foreign_origin_is_rejected( + local_api: tuple[AppRuntime, TestClient, Path], with_project_header: bool +) -> None: + runtime, client, workspace = local_api + http = cast(httpx.Client, client) + headers = {"origin": "http://evil.example"} + if with_project_header: + headers["x-thoth-project-id"] = "project:synthetic-upload" + before = _counts(runtime) + response = http.post( + "/files/stage", + headers=headers, + files={"file": ("synthetic.txt", b"synthetic only", "text/plain")}, + ) + files = [p for p in (workspace / "inbox").rglob("*") if p.is_file()] + assert response.status_code == 403 + assert response.json()["code"] == "LOCAL_ORIGIN_REJECTED" + assert files == [] + assert not _record( + RequestShape( + "file-header" if with_project_header else "file-no-header", + origin=headers["origin"], + content_type="multipart/form-data", + ), + "/files/stage", + "file-stage", + response, + before, + _counts(runtime), + executed=False, + staged_files=len(files), + custom_project_header=with_project_header, + ) diff --git a/tests/integration/test_m2_counter_cancel_boundary.py b/tests/integration/test_m2_counter_cancel_boundary.py index 1230b87..8a3b2b0 100644 --- a/tests/integration/test_m2_counter_cancel_boundary.py +++ b/tests/integration/test_m2_counter_cancel_boundary.py @@ -171,7 +171,7 @@ def links() -> list[str]: ) app = create_app(runtime.bus) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client: started = await client.post( "/rpc/async", diff --git a/tests/integration/test_migration_logging.py b/tests/integration/test_migration_logging.py new file mode 100644 index 0000000..8c85e11 --- /dev/null +++ b/tests/integration/test_migration_logging.py @@ -0,0 +1,33 @@ +"""Opening the runtime's database must not disable an already imported thoth logger.""" + +import io +import logging +from pathlib import Path + +from pytest import MonkeyPatch + +from thoth.adapters.storage.alembic_migrations import migrate_sqlite_database + + +def test_runtime_migration_keeps_preexisting_thoth_warning_enabled( + tmp_path: Path, monkeypatch: MonkeyPatch +) -> None: + monkeypatch.delenv("THOTH_ALEMBIC_URL", raising=False) + logger = logging.getLogger("thoth.migration_logging_regression") + assert logger.disabled is False + + migrate_sqlite_database(tmp_path / "workspace" / "db" / "thoth.sqlite3") + + assert logger.disabled is False + # fileConfig replaces root handlers. Attach a fresh observer after that normal + # configuration step; do not re-enable the logger or alter its effective level. + output = io.StringIO() + observer = logging.StreamHandler(output) + logger.addHandler(observer) + try: + logger.warning("THOTH_MIGRATION_LOGGER_SURVIVED") + assert output.getvalue() == "THOTH_MIGRATION_LOGGER_SURVIVED\n" + finally: + logger.removeHandler(observer) + observer.close() + output.close() diff --git a/tests/integration/test_prepare_synthetic_radar_demo.py b/tests/integration/test_prepare_synthetic_radar_demo.py index e93f6e2..b50e099 100644 --- a/tests/integration/test_prepare_synthetic_radar_demo.py +++ b/tests/integration/test_prepare_synthetic_radar_demo.py @@ -35,7 +35,7 @@ def _load() -> Any: @pytest.fixture def api(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Any]: monkeypatch.setenv("THOTH_WORKSPACE", str(tmp_path / "workspace")) - with TestClient(create_app()) as client: + with TestClient(create_app(), base_url="http://127.0.0.1:8765") as client: def send(path: str, body: bytes, headers: dict[str, str]) -> dict[str, Any]: response = client.post(path, content=body, headers=headers) diff --git a/tests/integration/test_public_browser_cleanup.py b/tests/integration/test_public_browser_cleanup.py new file mode 100644 index 0000000..5823849 --- /dev/null +++ b/tests/integration/test_public_browser_cleanup.py @@ -0,0 +1,158 @@ +"""Prove SDK force-close with this render's real, deliberately suspended Chromium.""" + +from __future__ import annotations + +import asyncio +import ctypes +import json +import os +import signal +import subprocess +from collections.abc import Awaitable, Callable +from time import monotonic +from typing import Any, cast + +import pytest +from playwright.async_api import Browser +from tests.integration.test_public_browser import SHELL, URI, FixtureReader + +from thoth.adapters.connectors.anonymous_browser import AnonymousChromiumReader +from thoth.adapters.connectors.anonymous_browser_lifecycle import AnonymousBrowserLifecycle +from thoth.adapters.connectors.common import utc_now +from thoth.domain.web_acquisition import WebPage + + +class OwnedProcess: + """Hold a Windows process handle so PID reuse cannot fool the exit assertion.""" + + def __init__(self, pid: int) -> None: + self.pid = pid + self.handle: Any = None + self.kernel: Any = None + self.native: Any = None + if os.name == "nt": + from ctypes import wintypes + + self.kernel = ctypes.WinDLL("kernel32", use_last_error=True) + self.kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + self.kernel.OpenProcess.restype = wintypes.HANDLE + self.kernel.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + self.kernel.CloseHandle.argtypes = [wintypes.HANDLE] + self.handle = self.kernel.OpenProcess(0x100000 | 0x1000 | 0x0800, False, pid) + assert self.handle, (pid, ctypes.get_last_error()) + self.native = ctypes.WinDLL("ntdll") + self.native.NtSuspendProcess.argtypes = [wintypes.HANDLE] + self.native.NtResumeProcess.argtypes = [wintypes.HANDLE] + + def alive(self) -> bool: + if os.name == "nt": + return self.kernel.WaitForSingleObject(self.handle, 0) == 258 + try: + os.kill(self.pid, 0) + except ProcessLookupError: + return False + return True + + def suspend(self) -> None: + if os.name == "nt": + assert self.native.NtSuspendProcess(self.handle) == 0 + else: + os.kill(self.pid, signal.SIGSTOP) + + def resume(self) -> None: + if not self.alive(): + return + if os.name == "nt": + assert self.native.NtResumeProcess(self.handle) == 0 + else: + os.kill(self.pid, signal.SIGCONT) + + def release(self) -> None: + if os.name == "nt": + self.kernel.CloseHandle(self.handle) + + +async def test_slow_close_preserves_page_and_force_stops_owned_processes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("DEBUG", "pw:browser") + launch = AnonymousBrowserLifecycle.launch + close = Browser.close + observed: dict[str, Any] = {} + processes: dict[int, OwnedProcess] = {} + browser_process: OwnedProcess | None = None + + async def observe_launch(self: AnonymousBrowserLifecycle, timeout: float) -> Browser: + nonlocal browser_process + browser = await launch(self, timeout) + # Private SDK access is confined to this proof, never used in product cleanup. + driver_pid = int(cast(Any, browser)._impl_obj._connection._transport._proc.pid) + session = await browser.new_browser_cdp_session() + send = cast(Callable[[str], Awaitable[dict[str, Any]]], session.send) + info = await send("SystemInfo.getProcessInfo") + await session.detach() + items = cast(list[dict[str, Any]], info["processInfo"]) + browser_pid = next(int(item["id"]) for item in items if item["type"] == "browser") + for pid in {driver_pid, *(int(item["id"]) for item in items)}: + processes[pid] = OwnedProcess(pid) + browser_process = processes[browser_pid] + stop: Callable[[], Awaitable[None]] | None = cast(Any, self)._stop_driver + assert stop is not None + observed.update(driver_pid=driver_pid, browser_pid=browser_pid, pids=sorted(processes)) + + async def observe_stop() -> None: + assert browser_process is not None + observed["browser_alive_before_force"] = browser_process.alive() + assert observed["browser_alive_before_force"] + started = monotonic() + try: + await stop() + finally: + observed["force_cleanup_seconds"] = monotonic() - started + observed["remaining_pids"] = [ + pid for pid, proc in processes.items() if proc.alive() + ] + + monkeypatch.setattr(self, "_stop_driver", observe_stop) + return browser + + async def suspend_then_close(self: Browser, *, reason: str | None = None) -> None: + assert browser_process is not None + browser_process.suspend() + observed["graceful_started"] = monotonic() + await close(self, reason=reason) + + monkeypatch.setattr(AnonymousBrowserLifecycle, "launch", observe_launch) + monkeypatch.setattr(Browser, "close", suspend_then_close) + page = WebPage( + requested_uri=URI, final_uri=URI, media_type="text/html", raw=SHELL, retrieved_at=utc_now() + ) + try: + result = await AnonymousChromiumReader(FixtureReader()).render( + page, max_bytes=20000, timeout=5 + ) + assert b"Measured latency: 12 ms" in result.raw + assert result.requested_uri == result.final_uri == URI + observed["cleanup_total_seconds"] = monotonic() - observed["graceful_started"] + assert observed["force_cleanup_seconds"] < 2 + assert observed["remaining_pids"] == [] + await asyncio.sleep(0) + finally: + observed["rescue_required"] = any(proc.alive() for proc in processes.values()) + print("BROWSER_CLEANUP " + json.dumps(observed, sort_keys=True)) + # A failing experiment must not leave its suspended browser behind. This + # rescue is not counted as proof of product cleanup; the test still fails. + if browser_process is not None and browser_process.alive(): + browser_process.resume() + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(browser_process.pid), "/T", "/F"], + check=False, + capture_output=True, + timeout=5, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + else: + os.killpg(browser_process.pid, signal.SIGKILL) + for proc in processes.values(): + proc.release() diff --git a/tests/integration/test_research_input_queue.py b/tests/integration/test_research_input_queue.py index fbb95af..abf2022 100644 --- a/tests/integration/test_research_input_queue.py +++ b/tests/integration/test_research_input_queue.py @@ -333,9 +333,7 @@ async def test_persisted_self_queue_projects_hold_without_rewriting_record( assert operation is not None item = handler.queue.read(operation) assert item is not None - handler.queue.store.save( - item.model_copy(update={"after_operation_id": item.operation_id}) - ) + handler.queue.store.save(item.model_copy(update={"after_operation_id": item.operation_id})) runtime.bus.close_tasks() await runtime.bus.drain() runtime.close() @@ -383,7 +381,7 @@ async def test_hosted_http_accepts_same_thread_queue_without_superseding_first( runtime = await setup(tmp_path, model, source=False) client = httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus)), - base_url="http://test", + base_url="http://127.0.0.1:8765", headers={"x-thoth-review-session": "queue-http-session"}, ) try: diff --git a/tests/integration/test_restore_access.py b/tests/integration/test_restore_access.py index 252b593..9bc0440 100644 --- a/tests/integration/test_restore_access.py +++ b/tests/integration/test_restore_access.py @@ -53,7 +53,7 @@ async def test_read_only_principal_gets_real_preview_without_writes_and_cannot_a ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=create_app(runtime.bus, auth=auth)), - base_url="http://test", + base_url="http://127.0.0.1:8765", ) as client: issued = await client.post( "/auth/session", diff --git a/tests/integration/test_source_cleanup_cancel_boundary.py b/tests/integration/test_source_cleanup_cancel_boundary.py index a94490d..8872315 100644 --- a/tests/integration/test_source_cleanup_cancel_boundary.py +++ b/tests/integration/test_source_cleanup_cancel_boundary.py @@ -140,7 +140,7 @@ def counts() -> tuple[int, ...]: monkeypatch.setattr(ControlRecordService, "create", create) app = create_app(runtime.bus) async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), base_url="http://test" + transport=httpx.ASGITransport(app=app), base_url="http://127.0.0.1:8765" ) as client: response = await client.post( "/rpc/async", diff --git a/tests/integration/test_workspace_runtime_exclusivity_baseline.py b/tests/integration/test_workspace_runtime_exclusivity_baseline.py index 95cb3a7..99f4a46 100644 --- a/tests/integration/test_workspace_runtime_exclusivity_baseline.py +++ b/tests/integration/test_workspace_runtime_exclusivity_baseline.py @@ -128,7 +128,9 @@ async def test_the_local_server_opens_its_runtime_on_the_first_request( application = create_app() transport = httpx.ASGITransport(app=application) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: assert (await client.get("/healthz")).status_code == 200 assert application.state.runtime is None assert not (workspace / "db").exists() diff --git a/tests/integration/test_workspace_runtime_lock.py b/tests/integration/test_workspace_runtime_lock.py index 88b1a8c..babb1ed 100644 --- a/tests/integration/test_workspace_runtime_lock.py +++ b/tests/integration/test_workspace_runtime_lock.py @@ -347,7 +347,9 @@ async def test_a_server_not_started_by_the_cli_answers_503_until_the_workspace_i } transport = httpx.ASGITransport(app=application) try: - async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: with _holder(tmp_path, workspace, _LOCK_HOLDER): refused = await client.post("/rpc", json=body) assert refused.status_code == 503 diff --git a/tests/integration/test_xai_workspace_route.py b/tests/integration/test_xai_workspace_route.py index 966fa4b..d4fb338 100644 --- a/tests/integration/test_xai_workspace_route.py +++ b/tests/integration/test_xai_workspace_route.py @@ -372,7 +372,9 @@ def rpc(method: str, key: str, values: dict[str, object]) -> dict[str, object]: } try: - async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + async with httpx.AsyncClient( + transport=transport, base_url="http://127.0.0.1:8765" + ) as client: response = await client.post("/rpc", json=start_request) assert response.status_code == 200, response.text started = response.json()["result"] @@ -540,19 +542,29 @@ def complete( assert replay == operation.result def reject( - self: LocalModelCredentials, provider: str, login_id: str, - response: str, auth_method: str, + self: LocalModelCredentials, + provider: str, + login_id: str, + response: str, + auth_method: str, ) -> dict[str, object]: del self, provider, login_id, auth_method raise ModelCredentialError(response) monkeypatch.setattr(LocalModelCredentials, "submit_login_response", reject) - rejected = await runtime.bus.dispatch(request( - "model/credential/login/complete", "manual-secret-error", - {"project_id": "system:workspace", "provider": "anthropic", - "auth_method": "claude_pkce", "login_id": "synthetic-attempt", - "response": secret}, - )) + rejected = await runtime.bus.dispatch( + request( + "model/credential/login/complete", + "manual-secret-error", + { + "project_id": "system:workspace", + "provider": "anthropic", + "auth_method": "claude_pkce", + "login_id": "synthetic-attempt", + "response": secret, + }, + ) + ) assert rejected.error is not None assert rejected.error.message == "MODEL_LOGIN_COMPLETE_FAILED" assert secret not in rejected.model_dump_json() diff --git a/tests/unit/services/test_authorization_clock_currentness.py b/tests/unit/services/test_authorization_clock_currentness.py new file mode 100644 index 0000000..eb1dff2 --- /dev/null +++ b/tests/unit/services/test_authorization_clock_currentness.py @@ -0,0 +1,108 @@ +"""Approval currentness follows append/lineage order, never wall-clock order.""" + +from collections.abc import Iterator +from datetime import timedelta + +import pytest +from sqlalchemy import create_engine +from tests.unit.services.test_authorization_currentness import NOW, STEP, envelope, plan + +from thoth.adapters.storage.action import SqliteActionStore +from thoth.adapters.storage.schema import action_authorizations +from thoth.application.services.authorization_currentness import approved_authorization +from thoth.domain.action_full import AuthorizationEnvelopeRecord + + +def approval() -> AuthorizationEnvelopeRecord: + return envelope(authorization_revision_id="approved", revision_digest="1" * 64) + + +def consumption(offset: int = -60) -> AuthorizationEnvelopeRecord: + return envelope( + authorization_revision_id="consumed", + revision_digest="2" * 64, + supersedes_revision_digest="1" * 64, + state="CONSUMED", + created_at=NOW + timedelta(seconds=offset), + consumed_at=NOW + timedelta(seconds=offset), + ) + + +@pytest.mark.parametrize("reverse", (False, True)) +def test_consumed_revision_prevents_reusing_approved_history(reverse: bool) -> None: + history = (approval(), consumption()) + if reverse: + history = tuple(reversed(history)) + assert approved_authorization(history, plan("a"), STEP) is None + + +def test_unique_latest_approved_is_selected_without_using_created_at() -> None: + first = envelope(authorization_revision_id="pending", revision_digest="0" * 64, state="PENDING") + latest = approval().model_copy( + update={ + "supersedes_revision_digest": first.revision_digest, + "created_at": NOW - timedelta(days=1), + } + ) + assert approved_authorization((latest, first), plan("a"), STEP) is latest + + +def test_another_authorization_can_approve_the_same_step() -> None: + fresh = envelope( + authorization_id="auth:2", authorization_revision_id="fresh", revision_digest="3" * 64 + ) + assert approved_authorization((approval(), consumption(), fresh), plan("a"), STEP) is fresh + + +def test_ambiguous_branch_does_not_select_an_old_or_sibling_approval() -> None: + sibling = envelope( + authorization_revision_id="sibling", + revision_digest="3" * 64, + supersedes_revision_digest="1" * 64, + ) + assert approved_authorization((approval(), consumption(), sibling), plan("a"), STEP) is None + + +@pytest.fixture +def store() -> Iterator[SqliteActionStore]: + engine = create_engine("sqlite+pysqlite:///:memory:") + action_authorizations.create(engine) + try: + yield SqliteActionStore(engine) + finally: + engine.dispose() + + +@pytest.mark.parametrize("offset", (-60, 0, 60)) +def test_store_selects_last_insert_even_on_rollback_or_tied_time( + store: SqliteActionStore, offset: int +) -> None: + first, last = approval(), consumption(offset) + store.add_authorization(first) + store.add_authorization(last) + current = store.read_authorization("p", "auth:1") + assert current is not None and current.revision_digest == last.revision_digest + assert current.created_at == last.created_at + assert [row.revision_digest for row in store.list_authorizations("p", "plan:1")] == [ + last.revision_digest + ] + fresh = envelope( + authorization_id="auth:2", authorization_revision_id="fresh", revision_digest="3" * 64 + ) + store.add_authorization(fresh) + assert [row.authorization_id for row in store.list_authorizations("p", "plan:1")] == [ + "auth:1", + "auth:2", + ] + + +def test_successor_query_is_scoped_and_does_not_depend_on_time(store: SqliteActionStore) -> None: + first = approval() + store.add_authorization(first) + assert not store.has_authorization_successor("p", "auth:1", first.revision_digest) + store.add_authorization(consumption()) + assert store.has_authorization_successor("p", "auth:1", first.revision_digest) + assert not store.has_authorization_successor("other-project", "auth:1", first.revision_digest) + assert not store.has_authorization_successor("p", "other-auth", first.revision_digest) + assert not store.has_authorization_successor("p", "auth:1", "9" * 64) + assert not store.has_authorization_successor("p", "auth:1", "2" * 64) diff --git a/tests/unit/test_anonymous_browser_lifecycle.py b/tests/unit/test_anonymous_browser_lifecycle.py new file mode 100644 index 0000000..329b31d --- /dev/null +++ b/tests/unit/test_anonymous_browser_lifecycle.py @@ -0,0 +1,201 @@ +"""Slow shutdown never turns a completed read into failure or hides caller cancellation.""" + +from __future__ import annotations + +import asyncio +from typing import Literal, cast + +import pytest +from playwright.async_api import Browser + +from thoth.adapters.connectors import anonymous_browser as rendering +from thoth.adapters.connectors.anonymous_browser_lifecycle import AnonymousBrowserLifecycle +from thoth.adapters.connectors.common import utc_now +from thoth.domain.connectors import ConnectorErrorCode, ConnectorFailure +from thoth.domain.web_acquisition import WebPage +from thoth.ports.web_reader import PublicReaderPort + + +class BrowserDouble: + def __init__(self, mode: str) -> None: + self.mode = mode + self.entered = asyncio.Event() + self.released = asyncio.Event() + self.finished = asyncio.Event() + + async def close(self) -> None: + self.entered.set() + try: + if self.mode == "slow": + await self.released.wait() + elif self.mode == "error": + raise RuntimeError("synthetic close failure") + elif self.mode == "cancel": + raise asyncio.CancelledError + finally: + self.finished.set() + + +class LifetimeDouble(AnonymousBrowserLifecycle): + def __init__(self, mode: str = "slow", *, stop_stalls: bool = False) -> None: + super().__init__() + self.fake = BrowserDouble(mode) + self.browser = cast(Browser, self.fake) + self.stopped = asyncio.Event() + self.stop_finished = asyncio.Event() + self.stop_stalls = stop_stalls + self.launch_error: Exception | None = None + self._stop_driver = self.stop + + async def launch(self, timeout: float) -> Browser: + if self.launch_error is not None: + self.browser = None + raise self.launch_error + return cast(Browser, self.fake) + + async def stop(self) -> None: + self.stopped.set() + try: + if self.stop_stalls: + await asyncio.Event().wait() + self.fake.released.set() + finally: + self.stop_finished.set() + + +class Renderer(rendering.AnonymousChromiumReader): + def __init__(self, mode: Literal["return", "fail", "wait"] = "return") -> None: + super().__init__(cast(PublicReaderPort, object())) + self.mode = mode + self.read_started = asyncio.Event() + self.read_error = ValueError("synthetic read failure") + + async def _render( + self, page: WebPage, browser: Browser, *, max_bytes: int, timeout: float + ) -> WebPage: + self.read_started.set() + if self.mode == "fail": + raise self.read_error + if self.mode == "wait": + await asyncio.Event().wait() + return page + + +def source() -> WebPage: + return WebPage( + requested_uri="https://fixture.example.org/read", + final_uri="https://fixture.example.org/read", + media_type="text/html", + raw=b"

Captured evidence

", + retrieved_at=utc_now(), + ) + + +def install(monkeypatch: pytest.MonkeyPatch, lifetime: LifetimeDouble) -> None: + monkeypatch.setattr(rendering, "AnonymousBrowserLifecycle", lambda: lifetime) + + +@pytest.mark.parametrize("mode", ("slow", "error", "cancel", "normal")) +async def test_completed_page_survives_cleanup( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, mode: str +) -> None: + lifetime = LifetimeDouble(mode) + install(monkeypatch, lifetime) + page = source() + assert await Renderer().render(page, max_bytes=1000, timeout=1) is page + assert lifetime.stopped.is_set() and lifetime.stop_finished.is_set() + assert lifetime.fake.finished.is_set() + if mode == "slow": + assert "ANONYMOUS_BROWSER_CLOSE_DELAYED" in caplog.text + elif mode in {"error", "cancel"}: + assert "ANONYMOUS_BROWSER_CLOSE_FAILED" in caplog.text + + +@pytest.mark.parametrize("stop_stalls", (False, True)) +async def test_read_failure_is_not_overwritten( + monkeypatch: pytest.MonkeyPatch, stop_stalls: bool +) -> None: + lifetime = LifetimeDouble(stop_stalls=stop_stalls) + install(monkeypatch, lifetime) + reader = Renderer("fail") + with pytest.raises(ValueError) as caught: + await reader.render(source(), max_bytes=1000, timeout=1) + assert caught.value is reader.read_error + assert lifetime.stopped.is_set() + + +async def test_read_timeout_stays_failure(monkeypatch: pytest.MonkeyPatch) -> None: + lifetime = LifetimeDouble("normal") + install(monkeypatch, lifetime) + with pytest.raises(TimeoutError): + await Renderer("wait").render(source(), max_bytes=1000, timeout=0.01) + assert lifetime.stopped.is_set() + + +@pytest.mark.parametrize("during_cleanup", (False, True)) +async def test_outer_cancel_and_repeated_cancel_propagate_after_cleanup( + monkeypatch: pytest.MonkeyPatch, during_cleanup: bool +) -> None: + lifetime = LifetimeDouble() + install(monkeypatch, lifetime) + reader = Renderer("return" if during_cleanup else "wait") + task = asyncio.create_task(reader.render(source(), max_bytes=1000, timeout=1)) + await (lifetime.fake.entered if during_cleanup else reader.read_started).wait() + task.cancel() + await lifetime.fake.entered.wait() + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert lifetime.stop_finished.is_set() and lifetime.fake.finished.is_set() + + +async def test_partial_launch_failure_stops_driver(monkeypatch: pytest.MonkeyPatch) -> None: + lifetime = LifetimeDouble() + failure = ValueError("synthetic launch failure") + lifetime.launch_error = failure + install(monkeypatch, lifetime) + with pytest.raises(ValueError) as caught: + await Renderer().render(source(), max_bytes=1000, timeout=1) + assert caught.value is failure + assert lifetime.stop_finished.is_set() + + +@pytest.mark.parametrize("stop_fails", (False, True)) +async def test_outer_cancel_during_driver_stop_is_not_swallowed( + monkeypatch: pytest.MonkeyPatch, stop_fails: bool +) -> None: + lifetime = LifetimeDouble("normal") + entered, release = asyncio.Event(), asyncio.Event() + + async def stop() -> None: + entered.set() + await release.wait() + if stop_fails: + raise RuntimeError("synthetic stop failure") + await lifetime.stop() + + monkeypatch.setattr(lifetime, "_stop_driver", stop) + install(monkeypatch, lifetime) + task = asyncio.create_task(Renderer().render(source(), max_bytes=1000, timeout=1)) + await entered.wait() + task.cancel() + await asyncio.sleep(0) + task.cancel() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_unconfirmed_cleanup_is_not_silently_reported_as_success( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + lifetime = LifetimeDouble("normal", stop_stalls=True) + install(monkeypatch, lifetime) + with pytest.raises(ConnectorFailure) as caught: + await Renderer().render(source(), max_bytes=1000, timeout=1) + assert caught.value.code == ConnectorErrorCode.DRIVER_ERROR + assert "ANONYMOUS_BROWSER_CLEANUP_UNCONFIRMED" in str(caught.value) + assert "ANONYMOUS_BROWSER_CLEANUP_UNCONFIRMED" in caplog.text + await asyncio.sleep(0) + assert lifetime.stop_finished.is_set() diff --git a/tests/unit/test_local_request_guard_unit.py b/tests/unit/test_local_request_guard_unit.py new file mode 100644 index 0000000..89c69db --- /dev/null +++ b/tests/unit/test_local_request_guard_unit.py @@ -0,0 +1,232 @@ +"""Exact local authority matching and rejection before any ASGI body consumption.""" + +from __future__ import annotations + +import json + +import pytest +from starlette.types import Message, Receive, Scope, Send + +from thoth.adapters.http.local_request_guard import LocalRequestGuard, check_local_request + + +def headers( + host: str = "localhost", origin: str | None = None, media: str = "application/json" +) -> list[tuple[bytes, bytes]]: + values = [(b"host", host.encode("latin-1")), (b"content-type", media.encode())] + if origin is not None: + values.append((b"origin", origin.encode("latin-1"))) + return values + + +@pytest.mark.parametrize( + "host", ("localhost", "LOCALHOST:5173", "127.0.0.1", "127.0.0.1:8765", "[::1]", "[::1]:65535") +) +def test_allowed_host_names_and_arbitrary_valid_ports(host: str) -> None: + assert check_local_request("POST", "/rpc", headers(host)) is None + + +@pytest.mark.parametrize( + "host", + ( + "", + "testserver", + "localhost.evil.example", + "127.0.0.1.evil.example", + "::1", + "[0:0:0:0:0:0:0:1]", + "[::2]", + "[::1]evil", + "[::1]evil:80", + "localhost:", + "localhost:bad", + "localhost:65536", + "localhost/path", + "localhost?", + "localhost#", + "evil@localhost", + " localhost", + "local\thost", + "[::1", + "localhost\\evil", + ), +) +def test_unlisted_or_malformed_host_is_rejected(host: str) -> None: + rejection = check_local_request("GET", "/healthz", headers(host)) + assert rejection is not None and rejection.code == "LOCAL_HOST_REJECTED" + + +@pytest.mark.parametrize( + "origin", + ( + None, + "http://localhost", + "HTTP://LOCALHOST:5173", + "http://127.0.0.1:8765", + "http://[::1]:5173", + ), +) +def test_allowed_origins(origin: str | None) -> None: + assert check_local_request("POST", "/rpc", headers(origin=origin)) is None + + +@pytest.mark.parametrize( + "origin", + ( + "", + "null", + "https://localhost", + "http://localhost.evil.example", + "http://127.0.0.1.evil.example", + "http://evil@localhost", + "http://localhost/", + "http://localhost?q=x", + "http://localhost#", + "http://[::2]", + "http://[::1]evil", + "http://[::1]evil:80", + "http://localhost:bad", + "http://local\nhost", + ), +) +def test_unlisted_or_malformed_origin_is_rejected(origin: str) -> None: + rejection = check_local_request("POST", "/new-action", headers(origin=origin)) + assert rejection is not None and rejection.code == "LOCAL_ORIGIN_REJECTED" + + +@pytest.mark.parametrize( + "path", + ( + "/rpc", + "/rpc/", + "/rpc/query", + "/rpc/async", + "/operations/op:1/cancel", + "/operations/op:1/cancel/", + ), +) +@pytest.mark.parametrize( + "media", ("application/json", "APPLICATION/JSON", "application/json; charset=utf-8") +) +def test_json_paths_accept_case_and_parameters(path: str, media: str) -> None: + assert check_local_request("POST", path, headers(media=media)) is None + + +@pytest.mark.parametrize( + "media", + ( + "", + "text/plain", + "application/x-www-form-urlencoded", + "application/jsonp", + "application/ld+json", + ), +) +def test_non_json_media_rejected(media: str) -> None: + rejection = check_local_request("POST", "/rpc", headers(media=media)) + assert rejection is not None and rejection.status == 415 + + +@pytest.mark.parametrize( + "name,code", + ( + (b"host", "LOCAL_HOST_REJECTED"), + (b"origin", "LOCAL_ORIGIN_REJECTED"), + (b"content-type", "JSON_CONTENT_TYPE_REQUIRED"), + ), +) +def test_duplicate_or_missing_required_headers(name: bytes, code: str) -> None: + values = headers(origin="http://localhost") + duplicate = next(item for item in values if item[0] == name) + rejected = check_local_request("POST", "/rpc", [*values, duplicate]) + assert rejected is not None and rejected.code == code + if name != b"origin": + rejected = check_local_request("POST", "/rpc", [item for item in values if item[0] != name]) + assert rejected is not None and rejected.code == code + + +def test_rejection_order_and_fixed_bilingual_payload() -> None: + for host, origin, code in ( + ("evil.example", "null", "LOCAL_HOST_REJECTED"), + ("localhost", "null", "LOCAL_ORIGIN_REJECTED"), + ("localhost", None, "JSON_CONTENT_TYPE_REQUIRED"), + ): + rejected = check_local_request("POST", "/rpc", headers(host, origin, "text/plain")) + assert rejected is not None and rejected.code == code + payload = rejected.payload() + assert set(payload) == {"code", "message_ko", "message_en"} + assert payload["message_ko"] and payload["message_en"] + assert "evil.example" not in json.dumps(payload) + + +@pytest.mark.parametrize("method", ("GET", "HEAD", "OPTIONS")) +def test_safe_methods_only_check_host(method: str) -> None: + assert check_local_request(method, "/rpc", headers(origin="null", media="")) is None + + +@pytest.mark.parametrize("method", ("POST", "PUT", "PATCH", "DELETE", "CUSTOM")) +def test_future_mutating_routes_automatically_check_origin(method: str) -> None: + rejected = check_local_request(method, "/future-route", headers(origin="null")) + assert rejected is not None and rejected.code == "LOCAL_ORIGIN_REJECTED" + assert check_local_request(method, "/files/stage", headers(media="multipart/form-data")) is None + + +@pytest.mark.parametrize("kind", ("http", "websocket")) +async def test_guard_rejects_before_body_or_downstream( + monkeypatch: pytest.MonkeyPatch, kind: str +) -> None: + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "LOCAL") + sent: list[Message] = [] + + async def downstream(scope: Scope, receive: Receive, send: Send) -> None: + pytest.fail("rejected request reached downstream") + + async def receive() -> Message: + pytest.fail("rejected request read body") + + async def send(message: Message) -> None: + sent.append(message) + + scope: Scope = { + "type": kind, + "method": "POST", + "path": "/rpc", + "headers": headers(origin="null"), + } + await LocalRequestGuard(downstream)(scope, receive, send) + if kind == "http": + assert sent[0]["status"] == 403 + assert not any(k.startswith(b"access-control-") for k, _ in sent[0]["headers"]) + payload = json.loads(sent[1]["body"]) + else: + assert sent[0]["type"] == "websocket.close" and sent[0]["code"] == 1008 + assert len(sent[0]["reason"].encode()) <= 123 + payload = json.loads(sent[0]["reason"]) + assert payload["code"] == "LOCAL_ORIGIN_REJECTED" + + +@pytest.mark.parametrize("kind,hosted", (("lifespan", False), ("http", True), ("websocket", True))) +async def test_other_scopes_and_hosted_pass_through( + monkeypatch: pytest.MonkeyPatch, kind: str, hosted: bool +) -> None: + monkeypatch.setenv("THOTH_DEPLOYMENT_MODE", "HOSTED_REVIEW" if hosted else "LOCAL") + reached: list[Scope] = [] + + async def downstream(scope: Scope, receive: Receive, send: Send) -> None: + reached.append(scope) + + async def receive() -> Message: + pytest.fail("unexpected receive") + + async def send(message: Message) -> None: + pytest.fail("unexpected rejection") + + scope: Scope = {"type": kind, "path": "/rpc", "headers": []} + await LocalRequestGuard(downstream)(scope, receive, send) + assert reached == [scope] + + +def test_websocket_checks_origin_even_on_get_but_does_not_require_json() -> None: + assert check_local_request("GET", "/socket", headers(media=""), websocket=True) is None + rejected = check_local_request("GET", "/socket", headers(origin="null"), websocket=True) + assert rejected is not None and rejected.code == "LOCAL_ORIGIN_REJECTED"