diff --git a/CHANGELOG.md b/CHANGELOG.md index 98aa8be..a43bc26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## Unreleased maintenance + +- Reject non-finite JSON values before CRUD/RPC writes even without a schema, and reject numeric column conversions that produce NaN/Infinity. Cache-disabled mutations no longer create unawaited placeholder coroutines. +- Encode idempotent operation replay results with the same JSON encoder as the first HTTP response, preserving binary text and numeric types; invalid JSON output rolls back before business changes commit. Stop collecting schema errors after the twenty-diagnostic response budget instead of materializing every invalid array element. +- Keep media and Editor attachment writes, flushes, output closure and publication off the event loop; drain owned work under repeated cancellation. Failed uploads remove partial objects, while completed metadata transactions retain their matching files if the caller disconnects. +- Negotiate identity encoding for bounded HTTP sources/JWKS and reject encoded bodies before decoding; ignore non-finite Retry-After values and close recovered request circuits. +- Reject unsafe checksum manifest records and linked/reparse-point source parents, pruning ignored build directories before exported-source traversal. +- Keep Editor document lock acquisition, validation, atomic publication and release in one IO worker. Cancelled saves stop waiting for busy locks and drain owned work before releasing the write mutex; slow disk work no longer blocks the event loop. +- Reload malformed cache envelopes instead of returning internal metadata or failing requests on invalid, non-finite or overflowing expiry values. +- Close Redis pub/sub resources when subscription setup fails or is cancelled, reject listeners that already stopped, and permit a clean subsequent connection attempt. +- These changes are separate from the frozen v0.5.2 publication candidates. + ## v0.5.2 (in development) - Normalize clipboard whitespace in release SHA input and add one-run publication of all four prepared component drafts, with full preflight, pinned source/asset manifests, controller CI gates and safe recovery after partial publication. diff --git a/MANIFEST.sha256 b/MANIFEST.sha256 index 9ae083b..330ecb5 100644 --- a/MANIFEST.sha256 +++ b/MANIFEST.sha256 @@ -20,7 +20,7 @@ d8ab139f74ef9ebe932cf11837c19b017b424a673a0b4c5f1349a8e41012fa7f .vscode/settin 759da1adabb4d60f41e3b678452af42192ee3a2c9cc40b0d30522153df374708 AUTHORS.md 699957fecf01059637d6d6531e82729aaa5a7e2425bef812f48552dafe7d97f7 BRANCHES.md db93badbb155f5e5d43633e044af661bb8d9b19e05a6a30607a4edb7964b22a0 BRAND.md -74d9b1ae7d651f6daddbb82ba7578455f6f25dca5476eb25ee7608f43bd734f5 CHANGELOG.md +8527c321a1f8a714c8b7ae36acde707f8a2eda8839fa0ef50863e3b14ea7ee7f CHANGELOG.md 75f224cc8d630a46c9673d36704453286a41504cc84742727fb0e8d129ed5f8a CITATION.cff ea87ab99ee7120f85aba29aa86365691dbeb12a1c15ace2b07bd4cbf90942af2 CODE_OF_CONDUCT.md d492cd4bc29540c5e21f11c313641d159974affd2dceef8d33f23ca82a2f809e CONTRIBUTING.md @@ -125,43 +125,44 @@ d9e2700d66d92e870781ce74d0814aa5a1370a68f2aef230a6de577ef28d7a80 docs/security/ e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 framework/__init__.py 37ab70ad90a853fc5d0a86d835ca2fa06f84080daabea7e598ca39bbd825812b framework/api_models.py 42cd8d1f79b0b8b1ef20e2c51ecbe050d10e062bb310b07d5ac21de3097796aa framework/audit.py -49f5a485c4fb070f5bc65e75c71ae7daa15c2652b55b112bc4ee6fec747fd24a framework/cache.py +95b9b126695b09d80eec99c21b32575e0d53b134b241e45b8a71930fc741b2d6 framework/cache.py e49b20a66f3b8a07308b58417eef7dbec8384bdaa6456ba46f8da0eee356fa64 framework/cli.py 5748add336febceec488dd06dd9fab28f8497bbd37f58c6398df974ab1119ef2 framework/config.py -31ecdde6bbb38403812cccf4914b9f552f6bb2761d84e3a5e21c09ba8d2e20b0 framework/crud.py +304224a7a39e7dc9597bffca4532ebd26cdf1fc302e483620613630a998f036c framework/crud.py 835ed21a963fb0f5b756949eeecec083848ac4f9b10b37e718bcb4c5ddba28b7 framework/datasources.py b5f28590f7d353bf8dbe2e5d1a7c6f807d8e3ded6bdc3350ed1c085400f07d79 framework/db.py 3e558e5ce7705bac95400393046a2387fdd07e3567935e1aec8bcaa79166e820 framework/doctor.py a540621a90e0ed13fa0ea32431d04c2e1df2f6a1cf91fed8d1183ed4addbd754 framework/domain/__init__.py 1124ac68ae374e94a7b87841fcfc8547b792cb85058a9f359370558c705337d5 framework/domain/packs.py -eb96703d2fe201b509d51ca56744150c8df1537c8453e52dfe3f138bb56eb74c framework/editor_api.py +93338736a1c0065daf090be673ab113eb04fe70d9a63f69cc2eb34ab8f1d3ef4 framework/editor_api.py 4b026114e194a4127f2bf8d3218234085f37bc4e713177703f0ab5ce46ac2feb framework/editor_call_client.py bec9cdc02bcc222a6bd7f0dc21c93553ef3b1fd640d2f3d7e777362dd7b1524d framework/editor_call_script.py 00cf05672dedd3f037ee6ced943b3640df9ded6976d6d97e3a974a28329ac651 framework/editor_database.py 9bea132ee2d88ab818c5a6752c2dfb4362006c73c2c58f7c770e5a0722e54ad4 framework/editor_identity.py -269d0ceb9b1535452453faa2754c3cb8570243cf6ab30711c50141418ce6a770 framework/events.py +2b817d7df232cf72cb072ed641205cd9cbaf3eed34cac7ebfd40fd30b20bb5f2 framework/events.py 5fe190becf047e082e54a08c66bd2980a2549aed2897fe1c4cacfbee724a4d37 framework/factory.py 626f782d6f296e1e631b911145ca8291ee8a09549c2e575d3386bc29291e7d91 framework/file_safety.py de12b36ae9bea9d38563bf47daf10f215321678afd963963451aab379053c8bf framework/hooks.py -04a2d8133281331f4951ee2cc15c9aefdd8331616f48519ddfb0a23fe6c3b1bc framework/media.py +dfefc7926f3efc62150101d07cc6eb3694c9cfc6419dc8cc756de34237a72dc3 framework/io_tasks.py +a43df3967576db8af37f56f930f818910c5d0458de26647cf63c2c317eb02662 framework/media.py 1cc63e515ff54980079194704b368d15f6b0c80ce6a85c186f45d6e9eea10630 framework/mongo.py 1570831cbee5aa560f97898992e14527bb5d9707d4cf0fc1591d93b9ff82bdae framework/observability.py -c8137f29f61005cf2bd2081fa9a86a7e1b3c420b9a43262ac9e731feaa3d09b6 framework/operations.py +2e1a7740561a043622f4532802fa6f569713fea9ef078f9cab99e000fbfe242a framework/operations.py 5ae44f680bdd7b9968ce1df3f9b537aa316998bbba6f5fbd0257818aa6df0c98 framework/protection.py 19a4b9a999a871274ef95a375d6e3b705ebd38267a8dffe2e16522f5d688ee11 framework/rate_limit.py 4175e816deef70b29791c2e5ed2734ae6de9532ece470c448fd4be9d99c49458 framework/readiness.py be55cabe848bf6c2bd6b45ba721766c5c3e85a013a0712d7da14a209ac975ed9 framework/reload.py 21be6652a548e7963b8a09ac0d5e7a5e0fe6eb4da13cc6d77b6553aac9d1c21e framework/responses.py 55df528b3b30dc303ce8e1d281e1971b6564ef450da70d5de66e92415d443733 framework/routers/__init__.py -1db8db23589e4b726d9f19366ec084e8725b370f6310abdf2dd26860a42d88ce framework/routers/project.py +261a18e1443cf6f3597494a4b4dd0070ba988c82fa2ba4709781f7be3946a4d6 framework/routers/project.py 5e2b98986d3920d7159995c69856de7a5fd27f58b567dd02704ea523738b12fd framework/runtime.py 0608b23a2ecf8ddc5364b1bde33f507187e4d93732327738af32d520999072bc framework/safe_regex.py 7a9f04faf18ed8ec3fe0c4a903b669e85d6e53d80b928804409131bdaa4aa743 framework/schema.py 84629be3c3d1aef5f3010f534b0566ff663d04dc3d88262e147a5aa2ae30e95d framework/security.py b29b44fb165c4af28ae91f6aac480cf2e09c0a0568a1972bb3ab3ecc2a4c31b3 framework/services/__init__.py -768657f21a7357358618788ff64c855f3ad56a83646262b48aa82b7f569cdcee framework/services/http_client.py +29334372b577f0e7239c748a0e3377ff0e4ca293b6783c403476cabe41a614ea framework/services/http_client.py 0ef95ba1273e79f8045a047021a925b60dcb7d885c3d50ab43c6a7af511ac3f5 framework/settings.py -c97250eddde42c7dae3534aca5bb2ccdbc4402b64f1b34ac127c194457c50042 framework/validation.py +e0c57d514017045d890da00f742947a46ca683caf12eb2c9338bae99aafc9930 framework/validation.py 697e2fba33660b8c58b2cde37f3a28dd243cdd12481b169e11a12287b9c836bc framework/wsgi.py 238447dd865450eabe163e23c3f8367468ecf8497c7e75a898999ad09d9796fc main.py dc5d3e115fb85134cfd39edec3fe7359ec8ec9bf7380197c521c5254425964d7 passenger_wsgi.py @@ -180,8 +181,8 @@ f2e6c2c9864842c26d4ba67e63204b1a2adc57bb88f7cc23a95e9d36091a6af6 run.py 90f44bdbc981a23b398611c5029b46962a122645c2d7dda3bda40852323f3ef4 schemas/manifest.schema.json 3b4610afd882c42db781765a285761452e172e96ddc003830725933f5b4d2b85 schemas/project.schema.json 2cb417944d83c33467932de605fa2c2a4adeeac58787026027bc09ce14273b80 scripts/benchmark_latency.py -ec63ea4fc896b86b3ea894b2c7dc172fd5024c9ca089b900fbd5fe54a31223e8 scripts/check_critical_coverage.py -14acf8bec2d2f24a34e80fb2fe6393638d02a7fe6fc54f638a576d87ee71e332 scripts/check_manifest.py +96ac71c5d6311831c42716cdea61957cabd37ba1a098f0af0ea5d38949da39f9 scripts/check_critical_coverage.py +bba005a9a513f11a4b07ff189676b5b00084b58efa87beeb926644969f7c7375 scripts/check_manifest.py 0b6cd71bcb38b64df0d0adc10aba8bb1506dcee315d1e7010e46a18d693433de scripts/generate_secret.py 95b9f1187cce1c37da17f7ea2cd5adc3b3a1916d76922bd824848156f124fe3d scripts/install-release.ps1 cde242339b86093a02a7e130311fbb764f3ab3404da49f9d3a08589765c6d6e0 scripts/install-release.sh @@ -213,9 +214,17 @@ ca8953b1d8d515c6b919b50d14941e74e454196a4532f5209785a27da2f023ed tests/fixtures 449ae8470143556c79adf7fe832b99a640dc0240ac932bfb3b5fc8d8c82b33a0 tests/fixtures/apps/__init__.py 1970df4c99aba901049f3fbf29f7b3043960922a31c0279398ee8414a37a3506 tests/test_cache.py c0c671e93ad23c24d1082780c1410f4b0219938239bcbf5e92c32e40db6a54f6 tests/test_config.py +6581736e9024206fde993e13ca3ce57c5787521c455318afb43488126d7073cf tests/test_editor_attachment_io.py 53103079f5d98b5627c98206164ea6188d01fa2f0b225105c0a619e3d043d985 tests/test_editor_call_browser.py +0fbbcd8b59ee47772b1ac8903fd7f9cdf080f15f67a8b53ec44c7e1a5418ebfe tests/test_editor_save_lifecycle.py 744494d80a5a9fd8b4d27f173301d6bdd22cc75dbc45e33514a2d8ae8021b78b tests/test_external_services.py +dc54c07db9e5ddea03cfdf5357825f45cfa6a922e79c160dfc97e1d005186a85 tests/test_http_response_bounds.py +dbca4b023f17f72104d284ce430fc36ef4ed9cb7efc8e846973accadc2f3b4d3 tests/test_idempotency_json_types.py 57d01ed3bbb98e0abd909458a90d607a2bcc9f0db7dcde635cff55617700c644 tests/test_integration_v04.py +6c93c65deb30fbdbfe75912af5beb8230dc6550791807f9b15169e5cb9bd4fb4 tests/test_manifest_boundaries.py +6a95cc65623ea7f6d1ae71c10bcb7a53b6996ae3f85c9586cd8894486ca19643 tests/test_media_io_lifecycle.py +3532b3bf30a0d9087e8883f460e0fd927c95c4a4e955adb9087ca408e416e2ea tests/test_night_resilience.py +38d857b87d48bb1c1edb542324023122b4f4e4f8ff80b717f86befcf9ec27eb7 tests/test_nonfinite_writes.py 1cb158b7430fdf903feb6b8af97189e4f22f44e20770ec019bcc0f6300db5134 tests/test_rate_limit.py fbdad48e2a0bc27edead5763efbfd73826e069ce1aba8ed89b8b4725f21927f0 tests/test_security.py 67bffeba2c2d79ffd51a29710655b9483a043e48c8c36bd5bf1724a1e6ed84dc tests/test_v03_features.py @@ -237,5 +246,6 @@ d17a269986f18a61d111f47f5db3004262834b75960024e1ce61d39a3f601d19 tests/test_v05 6ecf9f91d80018a89ef35f33dba23db9bbab0637b2e91a1020aa21c48cee9461 tests/test_v052_code_scanning.py b8a449a59138f6171f97a33b6a17fbee5b037a5ece50a05583443363d2e6679a tests/test_v052_latency.py 908e0c3f8a809e620cc66dd796865fab9c04c6bd8659674905a724cef857ebe8 tests/test_v052_publish_suite.py -0abe5dddbb0d9bd85b4229a2cb046956b1a0dcfa474d3cb4ae26354e0a95ecf6 tests/test_v052_regressions.py +1b9c0d8c3939aed7c8c966dd581c13e4287273ebc64ce61ebfc3348a8831f4eb tests/test_v052_regressions.py 892c76c9a6320ad52fa3c286a428fa1d08c8b403b4b2ee1ff4eb89db79986d1b tests/test_v052_release.py +c8a4ebc7ccbe2fb92f3fa317ab6f65fd0358fdb39c1220bfc80bf0bfdbdc05d8 tests/test_validation_error_budget.py diff --git a/framework/cache.py b/framework/cache.py index 3da7ad8..4400a03 100644 --- a/framework/cache.py +++ b/framework/cache.py @@ -3,6 +3,7 @@ import asyncio import json import logging +import math import time from collections import OrderedDict from hashlib import sha256 @@ -208,15 +209,22 @@ async def get_json_state(self, key: str) -> tuple[Any | None, str]: return None, "miss" try: parsed = json.loads(raw) - except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError): + envelope = isinstance(parsed, dict) and parsed.get("__forge_cache_v") == 1 + if envelope: + if "value" not in parsed: + raise ValueError("Cache envelope has no value") + fresh_until = parsed.get("fresh_until") + if isinstance(fresh_until, bool) or not isinstance(fresh_until, (int, float)) or not math.isfinite(fresh_until): + raise ValueError("Cache envelope has an invalid expiry") + except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError, OverflowError): log.warning("Discarding malformed cache entry key=%s", key) try: await self.backend.delete(key) except Exception: log.warning("Failed to delete malformed cache entry key=%s", key, exc_info=True) return None, "miss" - if isinstance(parsed, dict) and parsed.get("__forge_cache_v") == 1 and "value" in parsed: - state = "fresh" if time.time() <= float(parsed.get("fresh_until", 0)) else "stale" + if envelope: + state = "fresh" if time.time() <= fresh_until else "stale" return parsed["value"], state return parsed, "fresh" diff --git a/framework/crud.py b/framework/crud.py index feeac47..7c09b4c 100644 --- a/framework/crud.py +++ b/framework/crud.py @@ -2,12 +2,14 @@ import base64 import json +import math from datetime import date, datetime, timezone +from decimal import Decimal, InvalidOperation from itertools import groupby from typing import Any, Literal from fastapi import HTTPException, Request -from sqlalchemy import DateTime, and_, delete, func, insert, or_, select, update +from sqlalchemy import DateTime, Float, Numeric, and_, delete, func, insert, or_, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.sql.schema import Table @@ -44,6 +46,7 @@ def _write_payload( *, schema=None, mode: Literal["create", "patch", "replace"] = "patch", + dialect=None, ) -> dict[str, Any]: if not isinstance(payload, dict): raise HTTPException(status_code=422, detail="Resource payload must be a JSON object") @@ -71,12 +74,16 @@ def _write_payload( elif not has_default: raise HTTPException(status_code=422, detail={"missing_required_fields": [name]}) for name, value in list(data.items()): + if isinstance(table.c[name].type, (Float, Numeric)): + # Preserve the existing driver conversion of numeric strings, but + # reject values that would become NULL/Infinity after committing. + _coerce_for_column(table.c[name], value, dialect=dialect) if isinstance(value, str) and isinstance(table.c[name].type, DateTime): data[name] = _coerce_for_column(table.c[name], value) return data -def _coerce_for_column(column, value: Any): +def _coerce_for_column(column, value: Any, *, dialect=None): if value is None: return None try: @@ -103,8 +110,18 @@ def _coerce_for_column(column, value: Any): elif parsed.tzinfo is not None and not getattr(column.type, "timezone", False): parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None) return parsed - return pytype(value) - except (TypeError, ValueError) as exc: + converted = pytype(value) + if (isinstance(converted, float) and not math.isfinite(converted)) or ( + isinstance(converted, Decimal) and not converted.is_finite() + ): + raise ValueError("Numeric value must be finite") + if dialect is not None and isinstance(column.type, (Float, Numeric)): + processor = column.type.dialect_impl(dialect).bind_processor(dialect) + bound = processor(value) if processor is not None else converted + if (isinstance(bound, float) and not math.isfinite(bound)) or (isinstance(bound, Decimal) and not bound.is_finite()): + raise ValueError("Numeric value exceeds the database driver range") + return converted + except (TypeError, ValueError, OverflowError, InvalidOperation) as exc: raise HTTPException(status_code=422, detail=f"Invalid value for field {column.name}") from exc @@ -322,7 +339,7 @@ def _inject_policy_fields(data: dict[str, Any], resource: ResourceConfig, princi async def create_row(engine, table: Table, resource: ResourceConfig, principal: Principal, payload: dict[str, Any]): - data = _write_payload(payload, resource, table, schema=resource.create_schema, mode="create") + data = _write_payload(payload, resource, table, schema=resource.create_schema, mode="create", dialect=getattr(engine, "dialect", None)) _inject_policy_fields(data, resource, principal) try: async with engine.begin() as conn: @@ -342,7 +359,9 @@ async def batch_create_rows(engine, table: Table, resource: ResourceConfig, prin raise HTTPException(status_code=422, detail=f"Batch size must be 1..{resource.max_batch_size}") values = [] for payload in payloads: - data = _write_payload(payload, resource, table, schema=resource.create_schema, mode="create") + data = _write_payload( + payload, resource, table, schema=resource.create_schema, mode="create", dialect=getattr(engine, "dialect", None) + ) _inject_policy_fields(data, resource, principal) values.append(data) try: @@ -366,7 +385,14 @@ async def update_row( *, replace: bool = False, ): - data = _write_payload(payload, resource, table, schema=resource.update_schema, mode="replace" if replace else "patch") + data = _write_payload( + payload, + resource, + table, + schema=resource.update_schema, + mode="replace" if replace else "patch", + dialect=getattr(engine, "dialect", None), + ) if not data: raise HTTPException(status_code=422, detail="No writable fields supplied") pk = table.c[resource.primary_key] diff --git a/framework/editor_api.py b/framework/editor_api.py index 7e36c0e..ee6cab6 100644 --- a/framework/editor_api.py +++ b/framework/editor_api.py @@ -10,6 +10,7 @@ import re import shutil import tempfile +import threading import time import uuid from datetime import datetime @@ -27,6 +28,7 @@ from .editor_call_client import call_client_page, parse_ice_servers from .editor_database import browse_rows, database_catalog from .editor_identity import EDITOR_PERMISSIONS, EditorAccess, EditorIdentityStore, EditorPrincipal +from .io_tasks import finish_io, flush_file from .protection import client_ip, host_allowed, ip_allowed, request_is_https from .runtime import ProjectRuntime from .settings import Settings @@ -428,15 +430,21 @@ def _stage_project(self, project: Path, destination: Path) -> None: continue shutil.copyfile(source, target_dir / source.name, follow_symlinks=False) - async def _acquire_project_lock(self, project: Path) -> FileLock: + def _acquire_project_lock(self, project: Path, cancelled: threading.Event) -> FileLock | None: self._lock_dir.mkdir(mode=0o700, parents=True, exist_ok=True) name = hashlib.sha256(str(project).encode("utf-8")).hexdigest() + ".lock" - lock = FileLock(self._lock_dir / name, timeout=10) - try: - await asyncio.to_thread(lock.acquire) - except FileLockTimeout as exc: - raise HTTPException(status_code=503, detail="Project is busy; retry the save", headers={"Retry-After": "1"}) from exc - return lock + lock = FileLock(self._lock_dir / name) + deadline = time.monotonic() + 10 + while not cancelled.is_set(): + try: + lock.acquire(timeout=0) + return lock + except FileLockTimeout as exc: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise HTTPException(status_code=503, detail="Project is busy; retry the save", headers={"Retry-After": "1"}) from exc + cancelled.wait(min(0.05, remaining)) + return None def list_projects(self) -> list[dict]: if not self.apps_dir.exists(): @@ -512,6 +520,66 @@ def validate_project(self, project_name: str) -> dict: "data_sources": len(configured.data_sources), } + def _write_document_sync(self, project, target, normalized, data, payload, cancelled): + # FileLock keeps ownership per thread: acquire, stage, publish and release + # must stay in one worker, including validation/atomic-write failures. + lock = self._acquire_project_lock(project, cancelled) + if lock is None: + return + try: + if cancelled.is_set(): + return + self._assert_no_symlinks(project) + exists = target.is_file() and not target.is_symlink() + if exists: + current = target.read_bytes() + if payload.expected_sha256 != _digest(current): + raise HTTPException(status_code=409, detail="Document changed on the server; reload before saving") + elif payload.expected_sha256 != "new": + raise HTTPException(status_code=409, detail="Document no longer exists; reload before saving") + elif target.exists(): + raise HTTPException(status_code=409, detail="Document target is not a regular file") + + with tempfile.TemporaryDirectory(prefix="forge-editor-") as temp_root: + staged = Path(temp_root) / "project" + self._stage_project(project, staged) + staged_target, _ = _document_path( + staged, + normalized, + allow_hooks=self.settings.editor_allow_hooks, + allow_graphs=self.settings.editor_allow_graphs, + ) + staged_target.parent.mkdir(parents=True, exist_ok=True) + staged_target.write_bytes(data) + try: + _load_project_dir(staged, dotenv=dotenv_values(self.apps_dir.parent / ".env"), default_name=project.name) + except RuntimeError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + if cancelled.is_set(): + return + target.parent.mkdir(parents=True, exist_ok=True) + fd, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) + temporary = Path(temporary_name) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + if cancelled.is_set(): + return + os.replace(temporary, target) + if hasattr(os, "O_DIRECTORY"): + directory_fd = os.open(target.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + temporary.unlink(missing_ok=True) + finally: + lock.release() + async def write_document(self, project_name: str, raw_path: str, payload: DocumentWrite) -> str: if self.settings.editor_read_only: raise HTTPException(status_code=403, detail="Editor API is read-only") @@ -536,54 +604,27 @@ async def write_document(self, project_name: str, raw_path: str, payload: Docume _validate_graph_document(parsed) async with self._write_lock: - lock = await self._acquire_project_lock(project) + cancelled = threading.Event() + worker = asyncio.create_task( + asyncio.to_thread(self._write_document_sync, project, target, normalized, data, payload, cancelled), + name="forge-editor-save", + ) try: - self._assert_no_symlinks(project) - exists = target.is_file() and not target.is_symlink() - if exists: - current = target.read_bytes() - if payload.expected_sha256 != _digest(current): - raise HTTPException(status_code=409, detail="Document changed on the server; reload before saving") - elif payload.expected_sha256 != "new": - raise HTTPException(status_code=409, detail="Document no longer exists; reload before saving") - elif target.exists(): - raise HTTPException(status_code=409, detail="Document target is not a regular file") - - with tempfile.TemporaryDirectory(prefix="forge-editor-") as temp_root: - staged = Path(temp_root) / "project" - self._stage_project(project, staged) - staged_target, _ = _document_path( - staged, - normalized, - allow_hooks=self.settings.editor_allow_hooks, - allow_graphs=self.settings.editor_allow_graphs, - ) - staged_target.parent.mkdir(parents=True, exist_ok=True) - staged_target.write_bytes(data) + await asyncio.shield(worker) + except asyncio.CancelledError: + cancelled.set() + # Keep both the worker and write mutex owned until cleanup ends. + # Further caller cancellations must not strand an open lock/file. + while not worker.done(): try: - _load_project_dir(staged, dotenv=dotenv_values(self.apps_dir.parent / ".env"), default_name=project.name) - except RuntimeError as exc: - raise HTTPException(status_code=422, detail=str(exc)) from exc - - target.parent.mkdir(parents=True, exist_ok=True) - fd, temporary_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent) - temporary = Path(temporary_name) - try: - with os.fdopen(fd, "wb") as handle: - handle.write(data) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temporary, target) - if hasattr(os, "O_DIRECTORY"): - directory_fd = os.open(target.parent, os.O_RDONLY | os.O_DIRECTORY) - try: - os.fsync(directory_fd) - finally: - os.close(directory_fd) - finally: - temporary.unlink(missing_ok=True) - finally: - await asyncio.to_thread(lock.release) + await asyncio.shield(worker) + except asyncio.CancelledError: + continue + except Exception: + break + if not worker.cancelled(): + worker.exception() + raise digest = _digest(data) log.info("editor write project=%s document=%s client_policy=authorized sha256=%s", project.name, normalized, digest[:12]) return digest @@ -1135,36 +1176,58 @@ async def upload_attachment(area_id: str, request: Request, upload: UploadFile): temporary = Path(temporary_name) digest = hashlib.sha256() size = 0 + published = False + registered = False + + def publish(): + nonlocal published + os.replace(temporary, target) + published = True + + async def register(content_type): + nonlocal registered + metadata = await store_for(request).register_attachment( + access, + area_id=area_id, + original_name=raw_name, + stored_name=stored_name, + content_type=content_type, + size=size, + sha256=digest.hexdigest(), + ) + registered = True + return metadata + try: - with os.fdopen(fd, "wb") as handle: + try: + handle = os.fdopen(fd, "wb") + except BaseException: + os.close(fd) + raise + try: while chunk := await upload.read(1024 * 1024): size += len(chunk) if size > settings.editor_max_attachment_bytes: raise HTTPException(status_code=413, detail="Attachment exceeds the server policy limit") digest.update(chunk) - handle.write(chunk) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temporary, target) + await finish_io(asyncio.to_thread(handle.write, chunk)) + await finish_io(asyncio.to_thread(flush_file, handle)) + finally: + await finish_io(asyncio.to_thread(handle.close)) + await finish_io(upload.close()) + await finish_io(asyncio.to_thread(publish)) content_type = (upload.content_type or mimetypes.guess_type(raw_name)[0] or "application/octet-stream")[:160] if any(character in content_type for character in "\r\n\0"): content_type = "application/octet-stream" - try: - metadata = await store_for(request).register_attachment( - access, - area_id=area_id, - original_name=raw_name, - stored_name=stored_name, - content_type=content_type, - size=size, - sha256=digest.hexdigest(), - ) - except BaseException: - target.unlink(missing_ok=True) - raise + metadata = await finish_io(register(content_type)) finally: - temporary.unlink(missing_ok=True) - await upload.close() + try: + if published and not registered: + target.unlink(missing_ok=True) + finally: + temporary.unlink(missing_ok=True) + if not upload.file.closed: + await finish_io(upload.close()) await store_for(request).audit( access.principal, "attachment.create", diff --git a/framework/events.py b/framework/events.py index 4cb7cbe..8a6cf65 100644 --- a/framework/events.py +++ b/framework/events.py @@ -145,8 +145,9 @@ async def _ensure_listener(self, channel: str) -> None: ready_waiter = asyncio.create_task(ready.wait(), name=f"forge-redis-ready-{channel}") try: done, _ = await asyncio.wait({ready_waiter, task}, timeout=5.0, return_when=asyncio.FIRST_COMPLETED) - if task in done and not ready.is_set(): + if task in done: task.result() + raise RuntimeError(f"Redis event listener stopped for channel {channel!r}") if ready_waiter in done and ready_waiter.result(): return if ready.is_set(): @@ -159,9 +160,9 @@ async def _ensure_listener(self, channel: str) -> None: async def _listen(self, channel: str, ready: asyncio.Event) -> None: pubsub = self.redis.pubsub() - await pubsub.subscribe(self._redis_channel(channel)) - ready.set() try: + await pubsub.subscribe(self._redis_channel(channel)) + ready.set() async for message in pubsub.listen(): if message.get("type") != "message": continue @@ -173,7 +174,6 @@ async def _listen(self, channel: str, ready: asyncio.Event) -> None: except asyncio.CancelledError: raise finally: - ready.set() try: await pubsub.unsubscribe(self._redis_channel(channel)) finally: diff --git a/framework/io_tasks.py b/framework/io_tasks.py new file mode 100644 index 0000000..18d78a6 --- /dev/null +++ b/framework/io_tasks.py @@ -0,0 +1,27 @@ +"""Cancellation-safe ownership of blocking file work.""" + +import asyncio +import os + + +async def finish_io(awaitable): + """Retain IO ownership until completion, including repeated request cancellation.""" + task = asyncio.ensure_future(awaitable) + try: + return await asyncio.shield(task) + except asyncio.CancelledError: + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + continue + except BaseException: + break + if not task.cancelled(): + task.exception() + raise + + +def flush_file(out): + out.flush() + os.fsync(out.fileno()) diff --git a/framework/media.py b/framework/media.py index f84b7ad..862fc61 100644 --- a/framework/media.py +++ b/framework/media.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import base64 import hashlib import hmac @@ -16,6 +17,7 @@ from .config import MediaConfig from .file_safety import confined_parent, relative_parts +from .io_tasks import finish_io, flush_file from .security import media_table, media_usage_table log = logging.getLogger("json_api_forge.media") @@ -49,8 +51,29 @@ async def save_upload(self, upload: UploadFile, storage_key: str, max_bytes: int temporary = f".upload-{secrets.token_hex(16)}.tmp" target = temporary if parent_fd is not None else parent / temporary fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o600, dir_fd=parent_fd) + published = False + + def publish(): + nonlocal published + try: + if parent_fd is not None: + os.link(temporary, name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd, follow_symlinks=False) + else: + os.link(parent / temporary, parent / name) + except FileExistsError as exc: + raise HTTPException(409, "Media object already exists") from exc + published = True + if parent_fd is not None: + os.fsync(parent_fd) + try: - with os.fdopen(fd, "wb") as out: + try: + handle = os.fdopen(fd, "wb") + except BaseException: + os.close(fd) + raise + out = handle + try: while True: chunk = await upload.read(1024 * 1024) if not chunk: @@ -59,21 +82,23 @@ async def save_upload(self, upload: UploadFile, storage_key: str, max_bytes: int if total > max_bytes: raise HTTPException(status_code=413, detail="Media file exceeds configured maximum size") digest.update(chunk) - out.write(chunk) - out.flush() - os.fsync(out.fileno()) - try: - if parent_fd is not None: - os.link(temporary, name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd, follow_symlinks=False) - os.fsync(parent_fd) - else: - os.link(parent / temporary, parent / name) - except FileExistsError as exc: - raise HTTPException(409, "Media object already exists") from exc + await finish_io(asyncio.to_thread(out.write, chunk)) + await finish_io(asyncio.to_thread(flush_file, out)) + finally: + await finish_io(asyncio.to_thread(out.close)) + # Close the input before publication; cancelled input cleanup must + # never strand a published file without metadata. + await finish_io(upload.close()) + await finish_io(asyncio.to_thread(publish)) + except BaseException: + if published: + os.unlink(name if parent_fd is not None else parent / name, dir_fd=parent_fd) + raise finally: os.unlink(target, dir_fd=parent_fd) finally: - await upload.close() + if not upload.file.closed: + await finish_io(upload.close()) return total, digest.hexdigest() def delete(self, storage_key: str) -> None: @@ -148,7 +173,10 @@ async def save_media(*, engine, store: LocalMediaStore, project_slug: str, confi media_id = secrets.token_hex(16) storage_key = f"{project_slug}/{datetime.now(timezone.utc):%Y/%m}/{media_id}-{safe}" size, digest = await store.save_upload(upload, storage_key, config.max_upload_bytes) - try: + committed = False + + async def register(): + nonlocal committed if config.deduplicate: filt = (media_table.c.project_slug == project_slug) & (media_table.c.sha256 == digest) if config.deduplicate_scope == "owner": @@ -175,13 +203,18 @@ async def save_media(*, engine, store: LocalMediaStore, project_slug: str, confi if config.max_owner_bytes: await _reserve_usage(conn, project_slug, owner_subject, size, config.max_owner_bytes) await conn.execute(insert(media_table).values(**values)) + committed = True + return values + + try: + return await finish_io(register()) except BaseException: - try: - store.delete(storage_key) - except Exception: - log.exception("Failed to clean an uncommitted media object media_id=%s", media_id) + if not committed: + try: + store.delete(storage_key) + except Exception: + log.exception("Failed to clean an uncommitted media object media_id=%s", media_id) raise - return values def media_api_meta(meta: dict, *, include_owner: bool = False) -> dict: diff --git a/framework/operations.py b/framework/operations.py index 97ea35f..0dd15b8 100644 --- a/framework/operations.py +++ b/framework/operations.py @@ -8,6 +8,7 @@ from typing import Any from fastapi import HTTPException, Request +from fastapi.encoders import jsonable_encoder from sqlalchemy import ( Column, DateTime, @@ -312,7 +313,9 @@ async def execute_idempotent_operation( ) ) result = await _execute_on_connection(conn, operation, body=body, request=request, principal=principal) - payload = json.dumps(result, separators=(",", ":"), ensure_ascii=False, default=str) + # Store the same JSON representation FastAPI sends on the first + # response; repr() changes bytes and Decimal values on replay. + payload = json.dumps(jsonable_encoder(result), separators=(",", ":"), ensure_ascii=False, allow_nan=False) if len(payload.encode()) > operation.idempotency_max_response_bytes: raise HTTPException( status_code=413, diff --git a/framework/routers/project.py b/framework/routers/project.py index 09c0660..6f9a784 100644 --- a/framework/routers/project.py +++ b/framework/routers/project.py @@ -347,7 +347,8 @@ async def create_handler( p = await principal_for(request, _rt) require(p, _r.permissions.get("create", f"{_r.path}.create")) value = await create_row(_rt.registry.engines[_r.database], _rt.registry.tables[_tk], _r, p, payload) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value add_route( @@ -366,7 +367,8 @@ async def batch_create_handler( p = await principal_for(request, _rt) require(p, _r.permissions.get("create", f"{_r.path}.create")) value = await batch_create_rows(_rt.registry.engines[_r.database], _rt.registry.tables[_tk], _r, p, payload) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value add_route( @@ -387,7 +389,8 @@ async def patch_handler( value = await update_row( _rt.registry.engines[_r.database], _rt.registry.tables[_tk], _r, p, item_id, payload, replace=False ) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value async def put_handler( @@ -396,7 +399,8 @@ async def put_handler( p = await principal_for(request, _rt) require(p, _r.permissions.get("update", f"{_r.path}.update")) value = await update_row(_rt.registry.engines[_r.database], _rt.registry.tables[_tk], _r, p, item_id, payload, replace=True) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value add_route( @@ -421,7 +425,8 @@ async def delete_handler(item_id: str, request: Request, _r=resource, _tk=table_ p = await principal_for(request, _rt) require(p, _r.permissions.get("delete", f"{_r.path}.delete")) value = await delete_row(_rt.registry.engines[_r.database], _rt.registry.tables[_tk], _r, p, item_id) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value add_route( @@ -787,7 +792,8 @@ async def data_create(request: Request, payload: dict[str, Any] = Body(...), _s= p = await principal_for(request, _rt) require(p, _s.write_permission or _s.permission) if not _s.public_write else None value = await _rt.data_sources.create(_s, payload) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value async def data_patch( @@ -796,21 +802,24 @@ async def data_patch( p = await principal_for(request, _rt) require(p, _s.write_permission or _s.permission) if not _s.public_write else None value = await _rt.data_sources.update(_s, item_id, payload, replace=False) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value async def data_put(item_id: str, request: Request, payload: dict[str, Any] = Body(...), _s=source, _rt=runtime, _ns=namespace): p = await principal_for(request, _rt) require(p, _s.write_permission or _s.permission) if not _s.public_write else None value = await _rt.data_sources.update(_s, item_id, payload, replace=True) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value async def data_delete(item_id: str, request: Request, _s=source, _rt=runtime, _ns=namespace): p = await principal_for(request, _rt) require(p, _s.write_permission or _s.permission) if not _s.public_write else None value = await _rt.data_sources.delete(_s, item_id) - await _rt.cache.invalidate_namespace(_ns) if _rt.cache else asyncio.sleep(0) + if _rt.cache: + await _rt.cache.invalidate_namespace(_ns) return value write_openapi = {"security": []} if source.public_write else None diff --git a/framework/services/http_client.py b/framework/services/http_client.py index 84c98f5..0a2bc0d 100644 --- a/framework/services/http_client.py +++ b/framework/services/http_client.py @@ -2,6 +2,7 @@ import asyncio import ipaddress +import math import socket import time from concurrent.futures import ThreadPoolExecutor @@ -145,16 +146,27 @@ def _retry_delay(attempt: int, response: httpx.Response | None = None) -> float: raw = response.headers.get("Retry-After") if raw: try: - return min(max(float(raw), 0.0), 10.0) + delay = float(raw) + if math.isfinite(delay): + return min(max(delay, 0.0), 10.0) except ValueError: pass return min(0.25 * (2**attempt), 2.0) async def _send_once(self, method: str, url: str, *, max_response_bytes: int | None = None, **kwargs) -> httpx.Response: + if max_response_bytes is not None: + # Decoders may allocate an entire expanded chunk before yielding it. + # A post-decode byte count therefore cannot bound compressed input. + headers = httpx.Headers(kwargs.get("headers")) + headers["Accept-Encoding"] = "identity" + kwargs["headers"] = headers request = self.client.build_request(method, url, **kwargs) response = await self.client.send(request, stream=True) try: if max_response_bytes is not None: + encoding = response.headers.get("content-encoding", "").strip().lower() + if encoding not in {"", "identity"}: + raise ResponseTooLarge("Bounded upstream responses require identity content encoding") raw_length = response.headers.get("content-length") if raw_length: try: @@ -227,8 +239,12 @@ async def request( try: response = await self._send_once(method, url, max_response_bytes=max_response_bytes, **kwargs) if response.status_code not in _RETRYABLE_STATUS: - response.raise_for_status() + # A successful retry (or an ordinary client error) proves + # the upstream is reachable again, even if an earlier + # attempt opened the circuit inside this request. + state.opened_at = None state.failures = 0 + response.raise_for_status() return response response.raise_for_status() state.failures = 0 diff --git a/framework/validation.py b/framework/validation.py index 43f7ce6..9992a4c 100644 --- a/framework/validation.py +++ b/framework/validation.py @@ -1,6 +1,7 @@ from __future__ import annotations import math +from itertools import islice from typing import Any from fastapi import HTTPException, Request @@ -71,11 +72,36 @@ def _additional_properties(validator, additional, instance, schema): ) +def _check_finite_json(instance: Any, label: str) -> None: + # Walk containers lazily so wide arrays use depth-bounded traversal state. + pending = [(iter(((None, instance),)), ())] + while pending: + values, path = pending[-1] + try: + key, value = next(values) + except StopIteration: + pending.pop() + continue + child_path = path if key is None else (*path, str(key)) + if isinstance(value, float) and not math.isfinite(value): + raise HTTPException( + status_code=422, + detail={"schema": label, "errors": [{"path": ".".join(child_path) or "$", "message": "JSON numbers must be finite"}]}, + ) + if isinstance(value, dict): + pending.append((iter(value.items()), child_path)) + elif isinstance(value, list): + pending.append((iter(enumerate(value)), child_path)) + + def validate_json_schema(instance: Any, schema: dict[str, Any] | None, *, label: str = "request") -> None: + _check_finite_json(instance, label) if not schema: return validator = LinearDraft202012Validator(schema) - errors = sorted(validator.iter_errors(instance), key=lambda e: list(e.absolute_path)) + # Only collect diagnostics that can be returned. Large invalid arrays must + # not allocate every ValidationError merely to discard all but twenty. + errors = sorted(islice(validator.iter_errors(instance), 20), key=lambda e: list(e.absolute_path)) if not errors: return details = [] diff --git a/scripts/check_critical_coverage.py b/scripts/check_critical_coverage.py index f639211..a9dea3c 100644 --- a/scripts/check_critical_coverage.py +++ b/scripts/check_critical_coverage.py @@ -15,6 +15,7 @@ "framework/db.py": 80.0, "framework/events.py": 80.0, "framework/factory.py": 80.0, + "framework/io_tasks.py": 80.0, "framework/media.py": 80.0, "framework/mongo.py": 80.0, "framework/observability.py": 80.0, diff --git a/scripts/check_manifest.py b/scripts/check_manifest.py index 7c71325..497563e 100644 --- a/scripts/check_manifest.py +++ b/scripts/check_manifest.py @@ -3,6 +3,9 @@ import argparse import hashlib +import os +import re +import stat import subprocess from pathlib import Path @@ -20,7 +23,7 @@ def digest(path: Path) -> str: def _release_path(relative: str) -> bool: parts = relative.split("/") - if not relative or any(part in {"", ".", ".."} for part in parts) or "\\" in relative: + if not relative or any(part in {"", ".", ".."} for part in parts) or "\\" in relative or ":" in relative: raise ValueError(f"Unsafe release path: {relative!r}") if any(ord(char) < 32 or ord(char) == 127 for char in relative): raise ValueError(f"Control character in release path: {relative!r}") @@ -51,6 +54,42 @@ def _release_path(relative: str) -> bool: ) +def _checked_file(relative: str) -> Path: + if not _release_path(relative): + raise ValueError(f"Ignored release path: {relative}") + path = ROOT / relative + current = path + while current != ROOT: + try: + info = current.lstat() + except FileNotFoundError: + pass + else: + if stat.S_ISLNK(info.st_mode) or getattr(info, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0): + raise ValueError(f"Release paths may not contain symbolic links or reparse points: {relative}") + current = current.parent + return path + + +def _exported_files() -> set[str]: + def failed_scan(exc): + raise exc + + candidates = set() + for directory, folders, names in os.walk(ROOT, followlinks=False, onerror=failed_scan): + allowed = [] + for name in folders: + relative = (Path(directory) / name).relative_to(ROOT).as_posix() + if _release_path(relative): + _checked_file(relative) + allowed.append(name) + # Reject linked parents before descending, including Windows junctions + # that is_symlink/followlinks=False alone do not cover. + folders[:] = allowed + candidates.update((Path(directory) / name).relative_to(ROOT).as_posix() for name in names) + return candidates + + def release_files() -> set[str]: try: repository = subprocess.check_output( @@ -68,14 +107,12 @@ def release_files() -> set[str]: ).decode() candidates = {value for value in raw.split(chr(0)) if value} except (OSError, subprocess.SubprocessError, UnicodeDecodeError, ValueError): - candidates = {path.relative_to(ROOT).as_posix() for path in ROOT.rglob("*")} + candidates = _exported_files() result = set() for relative in candidates: if not _release_path(relative): continue - path = ROOT / relative - if path.is_symlink(): - raise ValueError(f"Release files may not be symbolic links: {relative}") + path = _checked_file(relative) if path.is_file(): result.add(relative) return result @@ -104,6 +141,12 @@ def main(argv: list[str] | None = None) -> int: d, rel = line.split(" ", 1) except ValueError as exc: raise SystemExit(f"Malformed manifest line {n}") from exc + try: + if not re.fullmatch(r"[0-9a-f]{64}", d) or not _release_path(rel): + raise ValueError("Invalid digest or ignored release path") + _checked_file(rel) + except ValueError as exc: + raise SystemExit(f"Unsafe manifest line {n}: {exc}") from exc if rel in expected: raise SystemExit(f"Duplicate manifest entry: {rel}") expected[rel] = d diff --git a/tests/test_editor_attachment_io.py b/tests/test_editor_attachment_io.py new file mode 100644 index 0000000..1b8965b --- /dev/null +++ b/tests/test_editor_attachment_io.py @@ -0,0 +1,186 @@ +import asyncio +import threading +import time +from contextlib import asynccontextmanager + +import httpx +import pytest + +from framework import editor_api as editor +from tests.test_v050_editor_collaboration import SETUP_TOKEN, _account_app + + +@asynccontextmanager +async def connected_editor(tmp_path): + app, settings = _account_app(tmp_path) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as client: + response = await client.post( + "/__forge/editor/v1/setup/founder", + headers={"X-Forge-Setup-Token": SETUP_TOKEN}, + json={"username": "founder", "password": "Granite river orbits seven moons 42!", "display_name": "Founder"}, + ) + assert response.status_code == 201, response.text + client.headers["Authorization"] = "Bearer " + response.json()["access_token"] + response = await client.post("/__forge/editor/v1/areas", json={"project": "Notes", "name": "Attachments", "visibility": "open"}) + assert response.status_code == 201, response.text + area = response.json()["id"] + yield client, area, settings.editor_attachment_dir + + +@pytest.mark.asyncio +async def test_attachment_flush_and_output_close_do_not_block_requests(tmp_path, monkeypatch): + real_fsync = editor.os.fsync + real_fdopen = editor.os.fdopen + loop_thread = threading.get_ident() + disk_threads = [] + ticks = 0 + + def slow_fsync(fd): + disk_threads.append(threading.get_ident()) + time.sleep(0.08) + real_fsync(fd) + + class Output: + def __init__(self, *args, **kwargs): + self.file = real_fdopen(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self.file, name) + + def close(self): + disk_threads.append(threading.get_ident()) + time.sleep(0.08) + self.file.close() + + async def heartbeat(): + nonlocal ticks + while True: + ticks += 1 + await asyncio.sleep(0.005) + + async with connected_editor(tmp_path) as (client, area, root): + monkeypatch.setattr(editor.os, "fsync", slow_fsync) + monkeypatch.setattr(editor.os, "fdopen", Output) + pulse = asyncio.create_task(heartbeat()) + try: + response = await client.post(f"/__forge/editor/v1/areas/{area}/attachments", files={"upload": ("file.txt", b"payload")}) + finally: + pulse.cancel() + await asyncio.gather(pulse, return_exceptions=True) + assert response.status_code == 201, response.text + assert ticks >= 3 + assert disk_threads and all(thread != loop_thread for thread in disk_threads) + downloaded = await client.get("/__forge/editor/v1/attachments/" + response.json()["id"]) + assert downloaded.status_code == 200 and downloaded.content == b"payload" + assert len(list(root.glob("*.blob"))) == 1 + assert not list(root.glob(".upload-*")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stage", ["flush", "publish", "close"]) +async def test_cancelled_attachment_drains_disk_worker_before_cleanup(tmp_path, monkeypatch, stage): + started = threading.Event() + release = threading.Event() + finished = threading.Event() + real_fsync = editor.os.fsync + real_replace = editor.os.replace + real_fdopen = editor.os.fdopen + + def wait_for_cancel(): + started.set() + assert release.wait(3), "IO worker was not released" + finished.set() + + def slow_fsync(fd): + if stage == "flush": + wait_for_cancel() + real_fsync(fd) + + def slow_replace(*args, **kwargs): + real_replace(*args, **kwargs) + if stage == "publish": + wait_for_cancel() + + class Output: + def __init__(self, *args, **kwargs): + self.file = real_fdopen(*args, **kwargs) + + def __getattr__(self, name): + return getattr(self.file, name) + + def close(self): + if stage == "close": + wait_for_cancel() + self.file.close() + + async def wait_started(): + while not started.is_set(): + await asyncio.sleep(0.005) + + async with connected_editor(tmp_path) as (client, area, root): + monkeypatch.setattr(editor.os, "fsync", slow_fsync) + monkeypatch.setattr(editor.os, "replace", slow_replace) + monkeypatch.setattr(editor.os, "fdopen", Output) + task = asyncio.create_task(client.post(f"/__forge/editor/v1/areas/{area}/attachments", files={"upload": ("a.txt", b"payload")})) + try: + await asyncio.wait_for(wait_started(), timeout=4) + task.cancel() + await asyncio.sleep(0.01) + task.cancel() + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + assert finished.is_set() + assert not list(root.iterdir()) + response = await client.get(f"/__forge/editor/v1/areas/{area}/attachments") + assert response.status_code == 200 and response.json()["attachments"] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [False, True]) +async def test_cancelled_attachment_registration_preserves_only_committed_files(tmp_path, monkeypatch, failure): + registered = asyncio.Event() + release = asyncio.Event() + real_register = editor.EditorIdentityStore.register_attachment + + async def delayed_register(self, *args, **kwargs): + if failure: + registered.set() + await release.wait() + raise OSError("metadata storage failed") + result = await real_register(self, *args, **kwargs) + registered.set() + await release.wait() + return result + + async with connected_editor(tmp_path) as (client, area, root): + monkeypatch.setattr(editor.EditorIdentityStore, "register_attachment", delayed_register) + task = asyncio.create_task(client.post(f"/__forge/editor/v1/areas/{area}/attachments", files={"upload": ("a.txt", b"payload")})) + try: + await asyncio.wait_for(registered.wait(), timeout=4) + task.cancel() + await asyncio.sleep(0.01) + task.cancel() + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + response = await client.get(f"/__forge/editor/v1/areas/{area}/attachments") + assert response.status_code == 200 + attachments = response.json()["attachments"] + if failure: + assert attachments == [] and not list(root.iterdir()) + else: + assert len(attachments) == 1 + response = await client.get("/__forge/editor/v1/attachments/" + attachments[0]["id"]) + assert response.status_code == 200 and response.content == b"payload" + assert len(list(root.glob("*.blob"))) == 1 + assert not list(root.glob(".upload-*")) diff --git a/tests/test_editor_save_lifecycle.py b/tests/test_editor_save_lifecycle.py new file mode 100644 index 0000000..ea8704e --- /dev/null +++ b/tests/test_editor_save_lifecycle.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +import threading +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException +from filelock import FileLock + +import framework.editor_api as editor +from framework.editor_api import DocumentWrite, EditorControlPlane +from framework.settings import Settings + + +def control_for(tmp_path): + project = tmp_path / "app/Notes" + (project / "config").mkdir(parents=True) + (project / "app.json").write_text( + json.dumps({"slug": "notes", "name": "Notes", "databases": {"primary": {"url": "sqlite+aiosqlite:///:memory:"}}}), + encoding="utf-8", + ) + target = project / "config/resources.json" + target.write_text('{"resources": []}\n', encoding="utf-8") + control = EditorControlPlane(project.parent, Settings(_env_file=None)) + payload = DocumentWrite(content='{\n"resources": []\n}\n', expected_sha256=hashlib.sha256(target.read_bytes()).hexdigest()) + return control, project, target, payload + + +@pytest.mark.asyncio +async def test_save_locks_and_files_stay_in_the_same_worker(tmp_path, monkeypatch): + control, _, target, payload = control_for(tmp_path) + activity = [] + native = FileLock + + def lock_factory(*args, **kwargs): + lock = native(*args, **kwargs) + + class TrackingLock: + def acquire(self, *args, **kwargs): + activity.append(("acquire", threading.get_ident())) + return lock.acquire(*args, **kwargs) + + def release(self): + activity.append(("release", threading.get_ident())) + return lock.release() + + return TrackingLock() + + stage = control._stage_project + + def track_stage(*args): + activity.append(("stage", threading.get_ident())) + return stage(*args) + + monkeypatch.setattr(editor, "FileLock", lock_factory) + monkeypatch.setattr(control, "_stage_project", track_stage) + main_thread = threading.get_ident() + digest = await control.write_document("Notes", "config/resources.json", payload) + assert digest == hashlib.sha256(target.read_bytes()).hexdigest() + assert [name for name, _ in activity] == ["acquire", "stage", "release"] + assert len({thread for _, thread in activity}) == 1 + assert activity[0][1] != main_thread + # A second real save must reacquire rather than inherit a leaked file lock. + await control.write_document("Notes", "config/resources.json", DocumentWrite(content='{"resources": []}', expected_sha256=digest)) + + +@pytest.mark.asyncio +async def test_slow_staging_keeps_event_loop_responsive_and_cancel_drains_worker(tmp_path, monkeypatch): + control, _, target, payload = control_for(tmp_path) + before = target.read_bytes() + entered, release, finished = threading.Event(), threading.Event(), threading.Event() + stage = control._stage_project + + def slow_stage(*args): + entered.set() + try: + if not release.wait(2): + raise RuntimeError("Test stage was never released") + return stage(*args) + finally: + finished.set() + + monkeypatch.setattr(control, "_stage_project", slow_stage) + task = asyncio.create_task(control.write_document("Notes", "config/resources.json", payload)) + try: + assert await asyncio.to_thread(entered.wait, 1) + assert not finished.is_set(), "Staging blocked the loop until the worker finished" + task.cancel() + await asyncio.sleep(0.01) + assert not task.done() and control._write_lock.locked() + task.cancel() + await asyncio.sleep(0.01) + assert not task.done() and control._write_lock.locked() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert finished.is_set() and not control._write_lock.locked() + assert target.read_bytes() == before + assert not list(target.parent.glob(".*.tmp")) + monkeypatch.setattr(control, "_stage_project", stage) + await control.write_document("Notes", "config/resources.json", payload) + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_cancelled_lock_wait_does_not_acquire_a_lock_after_returning(tmp_path, monkeypatch): + control, project, _, payload = control_for(tmp_path) + control._lock_dir.mkdir() + path = control._lock_dir / (hashlib.sha256(str(project).encode()).hexdigest() + ".lock") + held = FileLock(path) + held.acquire() + attempted = threading.Event() + native = editor.FileLock + + def lock_factory(*args, **kwargs): + lock = native(*args, **kwargs) + + class TrackingLock: + def acquire(self, *args, **kwargs): + attempted.set() + return lock.acquire(*args, **kwargs) + + def release(self): + return lock.release() + + return TrackingLock() + + monkeypatch.setattr(editor, "FileLock", lock_factory) + task = asyncio.create_task(control.write_document("Notes", "config/resources.json", payload)) + try: + assert await asyncio.to_thread(attempted.wait, 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 1) + assert not control._write_lock.locked() + held.release() + # The cancelled worker must be fully gone before this next save begins. + await asyncio.wait_for(control.write_document("Notes", "config/resources.json", payload), 1) + finally: + held.release() + await asyncio.gather(task, return_exceptions=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("phase", ["stage", "replace"]) +async def test_save_failure_releases_lock_and_preserves_existing_document(tmp_path, monkeypatch, phase): + control, _, target, payload = control_for(tmp_path) + before = target.read_bytes() + + def fail(*args, **kwargs): + raise OSError("simulated disk failure") + + with monkeypatch.context() as patch: + if phase == "stage": + patch.setattr(control, "_stage_project", fail) + else: + patch.setattr(editor.os, "replace", fail) + with pytest.raises(OSError, match="disk failure"): + await control.write_document("Notes", "config/resources.json", payload) + assert target.read_bytes() == before + assert not list(target.parent.glob(".*.tmp")) + await asyncio.wait_for(control.write_document("Notes", "config/resources.json", payload), 1) + + +@pytest.mark.asyncio +async def test_project_busy_deadline_is_preserved(tmp_path, monkeypatch): + control, project, _, payload = control_for(tmp_path) + control._lock_dir.mkdir() + path = control._lock_dir / (hashlib.sha256(str(project).encode()).hexdigest() + ".lock") + held = FileLock(path) + held.acquire() + ticks = iter([0, 11]) + monkeypatch.setattr(editor, "time", SimpleNamespace(monotonic=lambda: next(ticks, 11))) + try: + # Direct worker invocation avoids altering the async loop's own clock. + with pytest.raises(HTTPException) as exc: + control._write_document_sync(project, target=None, normalized=None, data=None, payload=payload, cancelled=threading.Event()) + assert exc.value.status_code == 503 and exc.value.headers == {"Retry-After": "1"} + finally: + held.release() diff --git a/tests/test_http_response_bounds.py b/tests/test_http_response_bounds.py new file mode 100644 index 0000000..9420d5b --- /dev/null +++ b/tests/test_http_response_bounds.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import asyncio +import gzip + +import httpx +import pytest + +from framework.services.http_client import ResilientHTTPClient, ResponseTooLarge + + +class UnreadableBody(httpx.AsyncByteStream): + def __init__(self): + self.closed = False + + async def __aiter__(self): + raise AssertionError("Encoded body must be rejected before a decoder reads it") + yield b"" + + async def aclose(self): + self.closed = True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("encoding", ["gzip", "br", "deflate", "gzip, br", " GZip "]) +async def test_bounded_http_rejects_encoded_body_without_reading_it(encoding): + body = UnreadableBody() + attempts = [] + + async def handler(request): + attempts.append(request) + return httpx.Response(200, headers={"Content-Encoding": encoding}, stream=body) + + client = ResilientHTTPClient() + await client.client.aclose() + client.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + with pytest.raises(ResponseTooLarge, match="identity content encoding"): + await client.request("GET", "https://upstream.test/data", max_response_bytes=1024, retries=2) + assert body.closed and len(attempts) == 1 + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_identity_negotiation_preserves_caller_headers_without_mutating_them(): + headers = {"accept-encoding": "gzip", "X-Trace": "kept"} + + async def handler(request): + assert request.headers["accept-encoding"] == "identity" + assert request.headers["x-trace"] == "kept" + return httpx.Response(200, json={"ok": True}, headers={"Content-Encoding": "identity"}) + + client = ResilientHTTPClient() + await client.client.aclose() + client.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + assert (await client.request("GET", "https://upstream.test/data", headers=headers, max_response_bytes=1024)).json() == {"ok": True} + assert headers == {"accept-encoding": "gzip", "X-Trace": "kept"} + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_unbounded_explicit_client_call_preserves_existing_compressed_response_support(): + class Body(httpx.AsyncByteStream): + async def __aiter__(self): + yield gzip.compress(b'{"ok":true}') + + async def handler(request): + return httpx.Response(200, headers={"Content-Encoding": "gzip"}, stream=Body()) + + client = ResilientHTTPClient() + await client.client.aclose() + client.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + assert (await client.request("GET", "https://upstream.test/data")).json() == {"ok": True} + finally: + await client.close() + + +@pytest.mark.parametrize("raw", ["nan", "NaN", "inf", "-inf", "invalid"]) +def test_non_finite_retry_after_uses_bounded_backoff(raw): + response = httpx.Response(503, headers={"Retry-After": raw}) + assert ResilientHTTPClient._retry_delay(0, response) == 0.25 + + +@pytest.mark.asyncio +async def test_non_finite_retry_after_does_not_abort_a_retry(monkeypatch): + calls = [] + delays = [] + + async def handler(request): + calls.append(request) + return httpx.Response(503 if len(calls) == 1 else 200, json={"ok": True}, headers={"Retry-After": "nan"}) + + async def sleep(delay): + delays.append(delay) + + monkeypatch.setattr(asyncio, "sleep", sleep) + client = ResilientHTTPClient() + await client.client.aclose() + client.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + assert (await client.request("GET", "https://upstream.test/data", max_response_bytes=1024)).json() == {"ok": True} + assert len(calls) == 2 and delays == [0.25] + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_recovered_retry_closes_the_circuit_for_the_next_request(): + attempts = [] + + async def handler(request): + attempts.append(request) + return httpx.Response(503 if len(attempts) == 1 else 200, json={"ok": True}, headers={"Retry-After": "0"}) + + client = ResilientHTTPClient(failure_threshold=1) + await client.client.aclose() + client.client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + assert (await client.request("GET", "https://upstream.test/data", retries=1)).status_code == 200 + assert (await client.request("GET", "https://upstream.test/data", retries=0)).status_code == 200 + assert len(attempts) == 3 + finally: + await client.close() diff --git a/tests/test_idempotency_json_types.py b/tests/test_idempotency_json_types.py new file mode 100644 index 0000000..ea8954e --- /dev/null +++ b/tests/test_idempotency_json_types.py @@ -0,0 +1,109 @@ +from decimal import Decimal + +import pytest +from fastapi.encoders import jsonable_encoder +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine +from starlette.requests import Request + +from framework import operations +from framework.config import OperationConfig, SQLStatementConfig +from framework.security import Principal + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [b"hello", "paylaşım 👋".encode(), b""]) +async def test_sqlite_binary_replay_matches_first_wire_response(tmp_path, value): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'replay.db'}") + await operations.init_operation_idempotency(engine) + async with engine.begin() as connection: + await connection.execute(text("CREATE TABLE counts (value INTEGER)")) + await connection.execute(text("INSERT INTO counts VALUES (0)")) + operation = OperationConfig( + name="binary-result", + permission="rpc", + transaction=True, + idempotency=True, + statements=[ + SQLStatementConfig(sql="UPDATE counts SET value = value + 1", mode="execute"), + SQLStatementConfig(sql="SELECT :payload AS payload", mode="fetch_one", params={"payload": value}, result_name="record"), + ], + ) + request = Request({"type": "http", "method": "POST", "path": "/rpc", "query_string": b"", "headers": []}) + principal = Principal("api_key", "client", set(), set()) + try: + first, replayed = await operations.execute_idempotent_operation( + engine, project_slug="p", operation=operation, principal=principal, raw_key="same", body={}, request=request + ) + assert not replayed + second, replayed = await operations.execute_idempotent_operation( + engine, project_slug="p", operation=operation, principal=principal, raw_key="same", body={}, request=request + ) + assert replayed + assert jsonable_encoder(first) == jsonable_encoder(second) + assert second["record"]["payload"] == value.decode() + async with engine.connect() as connection: + assert (await connection.execute(text("SELECT value FROM counts"))).scalar_one() == 1 + finally: + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [Decimal("1.25"), Decimal("123")]) +async def test_decimal_driver_values_keep_json_number_type_on_replay(tmp_path, monkeypatch, value): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'replay.db'}") + await operations.init_operation_idempotency(engine) + + async def driver_result(*args, **kwargs): + return {"balance": value} + + monkeypatch.setattr(operations, "_execute_on_connection", driver_result) + operation = OperationConfig( + name="decimal-result", permission="rpc", transaction=True, idempotency=True, statements=[SQLStatementConfig(sql="SELECT 1")] + ) + request = Request({"type": "http", "method": "POST", "path": "/rpc", "query_string": b"", "headers": []}) + principal = Principal("api_key", "client", set(), set()) + try: + first, _ = await operations.execute_idempotent_operation( + engine, project_slug="p", operation=operation, principal=principal, raw_key="same", body={}, request=request + ) + second, replayed = await operations.execute_idempotent_operation( + engine, project_slug="p", operation=operation, principal=principal, raw_key="same", body={}, request=request + ) + assert replayed + assert jsonable_encoder(first) == second + assert isinstance(second["balance"], (float, int)) + finally: + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [b"\xff", float("inf"), float("-inf")]) +async def test_non_json_sql_results_roll_back_before_committing_replay(tmp_path, value): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'replay.db'}") + await operations.init_operation_idempotency(engine) + async with engine.begin() as connection: + await connection.execute(text("CREATE TABLE counts (value INTEGER)")) + await connection.execute(text("INSERT INTO counts VALUES (0)")) + operation = OperationConfig( + name="invalid-json-result", + permission="rpc", + transaction=True, + idempotency=True, + statements=[ + SQLStatementConfig(sql="UPDATE counts SET value = value + 1", mode="execute"), + SQLStatementConfig(sql="SELECT :payload", mode="scalar", params={"payload": value}, result_name="value"), + ], + ) + request = Request({"type": "http", "method": "POST", "path": "/rpc", "query_string": b"", "headers": []}) + principal = Principal("api_key", "client", set(), set()) + try: + with pytest.raises(ValueError): + await operations.execute_idempotent_operation( + engine, project_slug="p", operation=operation, principal=principal, raw_key="same", body={}, request=request + ) + async with engine.connect() as connection: + assert (await connection.execute(text("SELECT value FROM counts"))).scalar_one() == 0 + assert (await connection.execute(text("SELECT COUNT(*) FROM _forge_v4_operation_idempotency"))).scalar_one() == 0 + finally: + await engine.dispose() diff --git a/tests/test_manifest_boundaries.py b/tests/test_manifest_boundaries.py new file mode 100644 index 0000000..fd1e135 --- /dev/null +++ b/tests/test_manifest_boundaries.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import importlib.util +import os +import subprocess +from pathlib import Path + +import pytest + + +def checker(tmp_path, monkeypatch): + script = Path(__file__).resolve().parents[1] / "scripts/check_manifest.py" + spec = importlib.util.spec_from_file_location("manifest_boundary_check", script) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + root = tmp_path / "export" + root.mkdir() + monkeypatch.setattr(module, "ROOT", root) + monkeypatch.setattr(module, "MANIFEST", root / "MANIFEST.sha256") + return module, root + + +@pytest.mark.parametrize( + "relative,digest", + [ + ("../outside.txt", "a" * 64), + ("/outside.txt", "a" * 64), + ("C:/outside.txt", "a" * 64), + ("source:stream", "a" * 64), + ("folder\\outside.txt", "a" * 64), + (".env", "a" * 64), + ("build/private.txt", "a" * 64), + ("source.py", "g" * 64), + ("source.py", "abc"), + ], +) +def test_manifest_rejects_invalid_records_before_reading_file_content(tmp_path, monkeypatch, relative, digest): + module, root = checker(tmp_path, monkeypatch) + (root / "source.py").write_text("source", encoding="utf-8") + module.MANIFEST.write_text(f"{digest} {relative}\n", encoding="utf-8") + + def forbidden_read(_path): + raise AssertionError("Invalid records must not read or hash file content") + + monkeypatch.setattr(module, "digest", forbidden_read) + with pytest.raises(SystemExit, match="Unsafe manifest line"): + module.main([]) + + +def force_export_scan(module, monkeypatch): + def no_checkout(*args, **kwargs): + raise subprocess.CalledProcessError(128, "git") + + monkeypatch.setattr(module.subprocess, "check_output", no_checkout) + + +@pytest.mark.parametrize("kind", ["symlink", "junction"]) +def test_export_scan_rejects_external_linked_parents(tmp_path, monkeypatch, kind): + module, root = checker(tmp_path, monkeypatch) + outside = tmp_path / "outside" + outside.mkdir() + secret = outside / "private-key.txt" + secret.write_text("synthetic private fixture", encoding="utf-8") + link = root / "linked" + if kind == "junction": + if os.name != "nt": + pytest.skip("Windows junction test") + result = subprocess.run(["cmd.exe", "/c", "mklink", "/J", str(link), str(outside)], capture_output=True) + if result.returncode: + pytest.skip("Test filesystem cannot create a junction") + else: + try: + link.symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("Test filesystem cannot create a symlink") + try: + force_export_scan(module, monkeypatch) + with pytest.raises(ValueError, match="links or reparse points"): + module.release_files() + assert secret.read_text(encoding="utf-8") == "synthetic private fixture" + finally: + if kind == "junction": + link.rmdir() + else: + link.unlink() + + +def test_ignored_directory_links_are_pruned_from_export_scan(tmp_path, monkeypatch): + module, root = checker(tmp_path, monkeypatch) + outside = tmp_path / "outside" + outside.mkdir() + (outside / "private-key.txt").write_text("synthetic fixture", encoding="utf-8") + link = root / "build" + try: + link.symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("Test filesystem cannot create a symlink") + try: + (root / "source.py").write_text("source", encoding="utf-8") + force_export_scan(module, monkeypatch) + assert module.release_files() == {"source.py"} + finally: + link.unlink() + + +def test_valid_export_manifest_still_checks_file_changes_and_missing_files(tmp_path, monkeypatch): + module, root = checker(tmp_path, monkeypatch) + force_export_scan(module, monkeypatch) + source = root / "source.py" + source.write_text("source", encoding="utf-8") + module.write_manifest() + assert module.main([]) == 0 + source.write_text("changed", encoding="utf-8") + with pytest.raises(SystemExit, match="hash mismatch"): + module.main([]) + source.unlink() + with pytest.raises(SystemExit, match="missing"): + module.main([]) diff --git a/tests/test_media_io_lifecycle.py b/tests/test_media_io_lifecycle.py new file mode 100644 index 0000000..fff3a5b --- /dev/null +++ b/tests/test_media_io_lifecycle.py @@ -0,0 +1,261 @@ +import asyncio +import io +import threading +import time + +import pytest +from fastapi import UploadFile + +from framework import media +from framework.config import MediaConfig + + +@pytest.mark.asyncio +async def test_slow_media_flush_does_not_block_other_requests(tmp_path, monkeypatch): + store = media.LocalMediaStore(str(tmp_path)) + real_fsync = media.os.fsync + loop_thread = threading.get_ident() + flush_threads = [] + heartbeats = 0 + + def slow_fsync(fd): + flush_threads.append(threading.get_ident()) + time.sleep(0.12) + real_fsync(fd) + + async def heartbeat(): + nonlocal heartbeats + while True: + heartbeats += 1 + await asyncio.sleep(0.005) + + monkeypatch.setattr(media.os, "fsync", slow_fsync) + pulse = asyncio.create_task(heartbeat()) + upload = UploadFile(io.BytesIO(b"payload"), filename="a.bin") + try: + size, digest = await store.save_upload(upload, "app/a.bin", 100) + finally: + pulse.cancel() + await asyncio.gather(pulse, return_exceptions=True) + assert size == 7 and len(digest) == 64 + assert heartbeats >= 3 + assert flush_threads and all(thread != loop_thread for thread in flush_threads) + assert upload.file.closed + assert (tmp_path / "app/a.bin").read_bytes() == b"payload" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stage", ["flush", "publish", "read"]) +async def test_cancelled_upload_drains_io_and_removes_owned_files(tmp_path, monkeypatch, stage): + store = media.LocalMediaStore(str(tmp_path)) + started = threading.Event() + release = threading.Event() + finished = threading.Event() + real_fsync = media.os.fsync + real_link = media.os.link + calls = 0 + + def wait_for_cancel(): + started.set() + assert release.wait(3), "test did not release IO worker" + finished.set() + + def slow_fsync(fd): + nonlocal calls + calls += 1 + if stage == "flush" and calls == 1: + wait_for_cancel() + return real_fsync(fd) + + def slow_link(*args, **kwargs): + result = real_link(*args, **kwargs) + if stage == "publish": + wait_for_cancel() + return result + + class DelayedRead(UploadFile): + async def read(self, size=-1): + if stage == "read": + started.set() + await asyncio.Event().wait() + return await super().read(size) + + monkeypatch.setattr(media.os, "fsync", slow_fsync) + monkeypatch.setattr(media.os, "link", slow_link) + upload = DelayedRead(io.BytesIO(b"payload"), filename="a.bin") + task = asyncio.create_task(store.save_upload(upload, "app/a.bin", 100)) + + async def wait_started(): + while not started.is_set(): + await asyncio.sleep(0.005) + + try: + await asyncio.wait_for(wait_started(), timeout=4) + task.cancel() + await asyncio.sleep(0.01) + task.cancel() + if stage != "read": + assert not task.done(), "upload returned while its IO worker was still running" + release.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + assert stage == "read" or finished.is_set() + assert upload.file.closed + assert not (tmp_path / "app/a.bin").exists() + assert not list(tmp_path.rglob("*.tmp")) + + +@pytest.mark.asyncio +async def test_publication_durability_failure_removes_new_object(tmp_path, monkeypatch): + if media.os.name != "posix": + pytest.skip("directory descriptor fsync is a POSIX durability step") + real_fsync = media.os.fsync + calls = 0 + + def fail_directory_flush(fd): + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("directory flush failed") + return real_fsync(fd) + + monkeypatch.setattr(media.os, "fsync", fail_directory_flush) + upload = UploadFile(io.BytesIO(b"payload")) + store = media.LocalMediaStore(str(tmp_path)) + with pytest.raises(OSError, match="directory flush failed"): + await store.save_upload(upload, "app/a.bin", 100) + assert upload.file.closed + assert not (tmp_path / "app/a.bin").exists() + assert not list(tmp_path.rglob("*.tmp")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [False, True]) +async def test_media_output_close_runs_off_event_loop_even_on_write_failure(tmp_path, monkeypatch, failure): + native_fdopen = media.os.fdopen + loop_thread = threading.get_ident() + close_threads = [] + + class Output: + def __init__(self, *args, **kwargs): + self.file = native_fdopen(*args, **kwargs) + + def write(self, chunk): + if failure: + raise OSError("disk write failed") + return self.file.write(chunk) + + def flush(self): + return self.file.flush() + + def fileno(self): + return self.file.fileno() + + def close(self): + close_threads.append(threading.get_ident()) + time.sleep(0.03) + self.file.close() + + monkeypatch.setattr(media.os, "fdopen", Output) + upload = UploadFile(io.BytesIO(b"payload")) + store = media.LocalMediaStore(str(tmp_path)) + if failure: + with pytest.raises(OSError, match="disk write failed"): + await store.save_upload(upload, "app/a.bin", 100) + assert not (tmp_path / "app/a.bin").exists() + else: + await store.save_upload(upload, "app/a.bin", 100) + assert (tmp_path / "app/a.bin").read_bytes() == b"payload" + assert close_threads and all(thread != loop_thread for thread in close_threads) + assert upload.file.closed + assert not list(tmp_path.rglob("*.tmp")) + + +@pytest.mark.asyncio +async def test_repeated_cancellation_during_input_close_does_not_publish(tmp_path): + closing = asyncio.Event() + release = asyncio.Event() + + class SlowClose(UploadFile): + async def close(self): + closing.set() + await release.wait() + await super().close() + + store = media.LocalMediaStore(str(tmp_path)) + upload = SlowClose(io.BytesIO(b"payload")) + task = asyncio.create_task(store.save_upload(upload, "app/a.bin", 100)) + try: + await asyncio.wait_for(closing.wait(), timeout=3) + task.cancel() + await asyncio.sleep(0.01) + task.cancel() + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + assert upload.file.closed + assert not (tmp_path / "app/a.bin").exists() + assert not list(tmp_path.rglob("*.tmp")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [False, True]) +async def test_cancelled_media_registration_keeps_database_and_object_consistent(tmp_path, failure): + from contextlib import asynccontextmanager + + from sqlalchemy import select + from sqlalchemy.ext.asyncio import create_async_engine + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'metadata.db'}") + committed = asyncio.Event() + release = asyncio.Event() + + class DelayedEngine: + @asynccontextmanager + async def begin(self): + async with engine.begin() as connection: + yield connection + if failure: + committed.set() + await release.wait() + raise OSError("metadata commit failed") + committed.set() + await release.wait() + + async with engine.begin() as connection: + await connection.run_sync(media.media_table.metadata.create_all) + store = media.LocalMediaStore(str(tmp_path / "media")) + config = MediaConfig(deduplicate=False, max_owner_bytes=None, allowed_mime_types=[], allowed_extensions=[]) + upload = UploadFile(io.BytesIO(b"payload"), filename="a.bin") + task = asyncio.create_task( + media.save_media(engine=DelayedEngine(), store=store, project_slug="app", config=config, upload=upload, owner_subject="owner") + ) + try: + await asyncio.wait_for(committed.wait(), timeout=4) + task.cancel() + await asyncio.sleep(0.01) + task.cancel() + assert not task.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + async with engine.connect() as connection: + rows = (await connection.execute(select(media.media_table))).mappings().all() + objects = [path for path in store.root.rglob("*") if path.is_file()] + if failure: + assert rows == [] and objects == [] + else: + assert len(rows) == 1 and len(objects) == 1 + assert store.path_for(rows[0]["storage_key"]).read_bytes() == b"payload" + assert upload.file.closed + finally: + release.set() + await asyncio.gather(task, return_exceptions=True) + await engine.dispose() diff --git a/tests/test_night_resilience.py b/tests/test_night_resilience.py new file mode 100644 index 0000000..8997eac --- /dev/null +++ b/tests/test_night_resilience.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import asyncio +import json + +import pytest + +from framework.cache import CacheManager, MemoryTTLCache +from framework.events import RedisEventHub + + +@pytest.mark.asyncio +@pytest.mark.parametrize("expiry", [None, "broken", [], {}, True, float("nan"), float("inf"), float("-inf"), 10**400]) +async def test_corrupt_cache_expiry_reloads_instead_of_crashing(expiry): + backend = MemoryTTLCache() + manager = CacheManager(backend) + await backend.set("key", json.dumps({"__forge_cache_v": 1, "fresh_until": expiry, "value": "stale"}).encode(), 10) + + async def load(): + return {"healthy": True} + + try: + assert await manager.get_or_set_json("key", 10, load) == ({"healthy": True}, False) + assert await manager.get_or_set_json("key", 10, load) == ({"healthy": True}, True) + finally: + await manager.close() + + +@pytest.mark.asyncio +async def test_incomplete_cache_envelope_is_not_returned_to_the_caller(): + backend = MemoryTTLCache() + manager = CacheManager(backend) + try: + await backend.set("key", b'{"__forge_cache_v":1,"fresh_until":123}', 10) + assert await manager.get_json_state("key") == (None, "miss") + assert await backend.get("key") is None + finally: + await manager.close() + + +class PubSub: + def __init__(self, *, failure=None, terminate=False): + self.failure = failure + self.terminate = terminate + self.subscribed = asyncio.Event() + self.closed = False + self.unsubscribed = False + + async def subscribe(self, _channel): + if self.failure is not None: + raise self.failure + self.subscribed.set() + + async def listen(self): + if not self.terminate: + await asyncio.Event().wait() + if False: + yield {} + + async def unsubscribe(self, _channel): + self.unsubscribed = True + + async def aclose(self): + self.closed = True + + +class Redis: + def __init__(self, pubsub): + self.subscription = pubsub + self.published = [] + + def pubsub(self): + return self.subscription + + async def publish(self, *args): + self.published.append(args) + return 1 + + async def aclose(self): + pass + + +def hub_for(pubsub): + hub = RedisEventHub("redis://localhost", "test", "p") + hub.redis = Redis(pubsub) + return hub + + +@pytest.mark.asyncio +async def test_redis_subscribe_failure_closes_resources_and_allows_retry(): + pubsub = PubSub(failure=ConnectionError("disconnected")) + hub = hub_for(pubsub) + try: + with pytest.raises(ConnectionError, match="disconnected"): + await hub.publish("messages", {"message": "hello"}) + assert pubsub.closed and pubsub.unsubscribed + assert hub.redis.published == [] + healthy = PubSub() + hub.redis.subscription = healthy + assert await hub.publish("messages", {"message": "retry"}) == 1 + assert not healthy.closed + finally: + await hub.close() + assert healthy.closed + + +@pytest.mark.asyncio +async def test_redis_finished_listener_never_reports_successful_authorization(): + pubsub = PubSub(terminate=True) + hub = hub_for(pubsub) + try: + with pytest.raises(RuntimeError, match="listener stopped"): + await hub.publish("messages", {"message": "hello"}) + assert pubsub.closed and hub.redis.published == [] + finally: + await hub.close() + + +@pytest.mark.asyncio +async def test_cancelling_subscription_establishment_closes_resources(): + pubsub = PubSub() + started = asyncio.Event() + + async def stalled_subscribe(_channel): + started.set() + await asyncio.Event().wait() + + pubsub.subscribe = stalled_subscribe + hub = hub_for(pubsub) + task = asyncio.create_task(hub.publish("messages", {})) + try: + await started.wait() + await hub.close() + with pytest.raises(asyncio.CancelledError): + await task + assert pubsub.closed and pubsub.unsubscribed and hub.redis.published == [] + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) diff --git a/tests/test_nonfinite_writes.py b/tests/test_nonfinite_writes.py new file mode 100644 index 0000000..9ce0cbe --- /dev/null +++ b/tests/test_nonfinite_writes.py @@ -0,0 +1,161 @@ +import json +import sqlite3 +import warnings +from contextlib import closing + +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy import Column, Float, Numeric + +from framework.crud import _coerce_for_column +from framework.factory import create_app +from framework.settings import settings +from framework.validation import validate_json_schema + + +@pytest.fixture(params=[True, False, "decimal"], ids=["numeric-schema", "no-schema", "sqlite-decimal"]) +def numeric_app(tmp_path, monkeypatch, request): + project = tmp_path / "app/Numeric" + project.mkdir(parents=True) + database = tmp_path / "business.db" + schema = {"type": "object", "properties": {"value": {"type": "number", "minimum": 0}}} if request.param is True else None + config = { + "slug": "numeric", + "name": "Numeric", + "audit_enabled": False, + "databases": {"primary": {"url": f"sqlite+aiosqlite:///{database}"}}, + "security": {"bootstrap_enabled": True, "bootstrap_admin_key": "B" * 48, "jwt_enabled": False}, + "cache": {"enabled": False}, + "rate_limit": {"enabled": False}, + "resources": [ + { + "database": "primary", + "table": "numbers", + "path": "numbers", + "auto_create": True, + "batch_enabled": True, + "columns": { + "id": {"type": "integer", "primary_key": True, "nullable": False}, + "value": {"type": "float", "nullable": True}, + }, + "writable_fields": ["value"], + "create_schema": schema, + "update_schema": schema, + } + ], + "operations": [ + { + "name": "append-number", + "path": "rpc/append-number", + "permission": "*", + "transaction": True, + "input_schema": schema, + "statements": [{"sql": "INSERT INTO numbers (value) VALUES (:value)", "params": {"value": "$body.value"}}], + } + ], + } + if request.param == "decimal": + with closing(sqlite3.connect(database)) as connection: + connection.execute("CREATE TABLE numbers(id INTEGER PRIMARY KEY, value DECIMAL(1000,2))") + connection.commit() + config["resources"][0]["auto_create"] = False + config["resources"][0].pop("columns") + (project / "app.json").write_text(json.dumps(config), encoding="utf-8") + for name, value in { + "apps_dir": tmp_path / "app", + "internal_database_url": f"sqlite+aiosqlite:///{tmp_path / 'internal.db'}", + "jwt_secret": "J" * 64, + "bootstrap_admin_key": "", + "app_env": "development", + "redis_url": None, + }.items(): + monkeypatch.setattr(settings, name, value) + with TestClient(create_app(), raise_server_exceptions=False) as client: + response = client.post( + "/api/numeric/v1/admin/api-keys", headers={"X-API-Key": "B" * 48}, json={"permissions": ["*"], "name": "test"} + ) + assert response.status_code == 200, response.text + client.headers.update({"X-API-Key": response.json()["api_key"], "Content-Type": "application/json"}) + yield client, database + + +def test_nonfinite_payloads_never_commit_any_resource_or_rpc_write(numeric_app): + client, database = numeric_app + base = "/api/numeric/v1/" + response = client.post(base + "numbers", json={"value": 1.25}) + assert response.status_code == 200, response.text + row_id = response.json()["id"] + for token in ("NaN", "Infinity", "-Infinity", "1e999", '"NaN"', '"Infinity"', '"1e999"'): + body = '{"value":' + token + "}" + for method, endpoint, payload in [ + ("POST", "numbers", body), + ("PATCH", f"numbers/{row_id}", body), + ("PUT", f"numbers/{row_id}", body), + ("POST", "numbers/_batch", '[{"value":2.5},' + body + "]"), + ("POST", "rpc/append-number", body), + ]: + if endpoint == "rpc/append-number" and token.startswith('"'): + continue # A schema-free SQL RPC may legitimately bind text. + response = client.request(method, base + endpoint, content=payload) + assert response.status_code == 422, (method, endpoint, token, response.text) + with closing(sqlite3.connect(database)) as connection: + assert connection.execute("SELECT value FROM numbers").fetchall() == [(1.25,)] + + +def test_cache_disabled_successful_mutations_do_not_leak_coroutines(numeric_app): + client, _ = numeric_app + base = "/api/numeric/v1/numbers" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", RuntimeWarning) + response = client.post(base, json={"value": 1.25}) + assert response.status_code == 200, response.text + row_id = response.json()["id"] + assert client.patch(f"{base}/{row_id}", json={"value": 2.5}).status_code == 200 + assert client.put(f"{base}/{row_id}", json={"value": 3.5}).status_code == 200 + assert client.post(base + "/_batch", json=[{"value": 4.5}]).status_code == 200 + assert client.delete(f"{base}/{row_id}").status_code == 200 + assert not [warning for warning in caught if issubclass(warning.category, RuntimeWarning)] + + +@pytest.mark.parametrize("schema", [None, {}, {"type": "object"}]) +def test_nested_nonfinite_json_is_rejected_without_requiring_numeric_schema(schema): + with pytest.raises(HTTPException) as raised: + validate_json_schema({"nested": [1, {"amount": float("nan")}]}, schema) + assert raised.value.status_code == 422 + assert raised.value.detail["errors"][0]["path"] == "nested.1.amount" + + +@pytest.mark.parametrize("value", ["NaN", "Infinity", "-Infinity", "1e999", 10**400]) +def test_float_column_conversion_rejects_nonfinite_or_overflowing_values(value): + with pytest.raises(HTTPException) as raised: + _coerce_for_column(Column("value", Float), value) + assert raised.value.status_code == 422 + + +@pytest.mark.parametrize("value", ["NaN", "Infinity", "-Infinity"]) +def test_decimal_column_conversion_rejects_nonfinite_values(value): + with pytest.raises(HTTPException) as raised: + _coerce_for_column(Column("value", Numeric), value) + assert raised.value.status_code == 422 + + +@pytest.mark.parametrize("column_type", [Float, Numeric]) +def test_numeric_column_conversion_keeps_finite_strings_and_null(column_type): + column = Column("value", column_type) + assert _coerce_for_column(column, "1.25") == 1.25 + assert _coerce_for_column(column, None) is None + + +def test_numeric_range_validation_uses_actual_database_bind_policy(): + from decimal import Decimal + + from sqlalchemy.dialects.postgresql.asyncpg import PGDialect_asyncpg + from sqlalchemy.dialects.sqlite import dialect as SQLiteDialect + + column = Column("value", Numeric(1000, 2)) + with pytest.raises(HTTPException) as raised: + _coerce_for_column(column, "1e999", dialect=SQLiteDialect()) + assert raised.value.status_code == 422 + # Native Decimal binding must not inherit SQLite's float conversion limit. + assert _coerce_for_column(column, "1e999", dialect=PGDialect_asyncpg()) == Decimal("1e999") diff --git a/tests/test_v052_regressions.py b/tests/test_v052_regressions.py index 20820a9..50af2a5 100644 --- a/tests/test_v052_regressions.py +++ b/tests/test_v052_regressions.py @@ -261,7 +261,9 @@ async def run(): await client.client.aclose() client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle)) try: - response = await client.request("GET", "https://example.test", max_response_bytes=100) + # Unbounded explicit calls retain gzip decoding; bounded calls now + # negotiate identity and reject encodings before allocating output. + response = await client.request("GET", "https://example.test") assert response.json() == {"ok": True} assert "content-encoding" not in response.headers with pytest.raises(httpcore.ConnectTimeout): diff --git a/tests/test_validation_error_budget.py b/tests/test_validation_error_budget.py new file mode 100644 index 0000000..7d79f66 --- /dev/null +++ b/tests/test_validation_error_budget.py @@ -0,0 +1,36 @@ +import pytest +from fastapi import HTTPException + +from framework import validation + + +def test_large_invalid_array_stops_collecting_after_diagnostic_budget(monkeypatch): + original = validation.LinearDraft202012Validator + yielded = 0 + + class ObservedValidator: + def __init__(self, schema): + self.validator = original(schema) + + def iter_errors(self, instance): + nonlocal yielded + for error in self.validator.iter_errors(instance): + yielded += 1 + yield error + + monkeypatch.setattr(validation, "LinearDraft202012Validator", ObservedValidator) + with pytest.raises(HTTPException) as raised: + validation.validate_json_schema(["invalid"] * 10_000, {"type": "array", "items": {"type": "integer"}}) + assert raised.value.status_code == 422 + assert len(raised.value.detail["errors"]) == 20 + assert yielded <= 21 + assert raised.value.detail["errors"][0]["path"] == "0" + + +def test_error_budget_does_not_skip_valid_values_or_late_invalid_value(): + schema = {"type": "array", "items": {"type": "integer"}} + validation.validate_json_schema(list(range(4096)), schema) + values = list(range(4096)) + ["invalid"] + with pytest.raises(HTTPException) as raised: + validation.validate_json_schema(values, schema) + assert raised.value.detail["errors"][0]["path"] == "4096"