feat: validate contract-driven HTTP inputs

app/api/errors.py

@@ -11,6 +11,32 @@ from fastapi.responses import JSONResponse
 logger = logging.getLogger(__name__)
 
 
+class ApiError(Exception):
+    """A safe error intended for the Proxmox-compatible boundary."""
+
+    def __init__(
+        self, status_code: int, message: str, errors: dict[str, str] | None = None
+    ) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+        self.errors = errors
+
+
+class ContractValidationError(ApiError):
+    def __init__(self, errors: dict[str, str]) -> None:
+        super().__init__(400, "parameter verification failed", errors)
+
+
+async def api_error_handler(_request: Request, exc: Exception) -> JSONResponse:
+    if not isinstance(exc, ApiError):
+        raise TypeError("api_error_handler received an incompatible exception")
+    body: dict[str, Any] = {"data": None, "message": exc.message}
+    if exc.errors is not None:
+        body["errors"] = exc.errors
+    return JSONResponse(status_code=exc.status_code, content=body)
+
+
 async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
     """Log internal failures and return a stable non-FastAPI error envelope."""
 

app/api/registry.py

@@ -5,10 +5,12 @@ from __future__ import annotations
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 from typing import Any, Literal
+from urllib.parse import parse_qsl
 
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
 
+from app.api.errors import ContractValidationError
 from app.contracts.model import Method, Schema, Snapshot
 
 Handler = Callable[[Request, dict[str, Any]], Awaitable[Any]]
@@ -75,10 +77,7 @@ def _endpoint(
 ) -> Callable[[Request], Awaitable[JSONResponse]]:
     async def dispatch(request: Request) -> JSONResponse:
         handler = handlers.get(semantic_path, method.verb)
-        inputs = {
+        inputs = await _parse_inputs(request, method)
-            "path": dict(request.path_params),
-            "query": dict(request.query_params),
-        }
         if handler is not None:
             data = await handler(request, inputs)
         elif fallback == "schema-default":
@@ -97,6 +96,78 @@ def _endpoint(
     return dispatch
 
 
+async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]:
+    supplied: dict[str, Any] = dict(request.query_params)
+    supplied.update(request.path_params)
+    if request.method not in {"GET", "DELETE"}:
+        content_type = request.headers.get("content-type", "").split(";", 1)[0].strip()
+        if content_type == "application/json":
+            try:
+                body = await request.json()
+            except ValueError as exc:
+                raise ContractValidationError({"body": "invalid JSON"}) from exc
+            if not isinstance(body, dict):
+                raise ContractValidationError({"body": "expected an object"})
+            supplied.update(body)
+        elif content_type == "application/x-www-form-urlencoded":
+            supplied.update(dict(parse_qsl((await request.body()).decode())))
+
+    definitions = {parameter.name: parameter.definition for parameter in method.parameters}
+    errors: dict[str, str] = {}
+    parsed: dict[str, Any] = {}
+    for name, definition in definitions.items():
+        if name not in supplied:
+            if definition.optional:
+                if definition.default is not None:
+                    parsed[name] = definition.default
+                continue
+            errors[name] = "property is missing and it is not optional"
+            continue
+        try:
+            parsed[name] = _coerce(supplied[name], definition)
+        except (TypeError, ValueError) as exc:
+            errors[name] = str(exc)
+    for name in supplied.keys() - definitions.keys():
+        if name not in request.path_params:
+            errors[name] = "property is not defined in schema"
+    if errors:
+        raise ContractValidationError(dict(sorted(errors.items())))
+    return {"values": parsed, "path": dict(request.path_params)}
+
+
+def _coerce(value: Any, schema: Schema) -> Any:
+    if schema.type == "integer":
+        parsed: Any = int(value)
+    elif schema.type == "number":
+        parsed = float(value)
+    elif schema.type == "boolean":
+        if isinstance(value, bool):
+            parsed = value
+        elif str(value).lower() in {"1", "true", "yes", "on"}:
+            parsed = True
+        elif str(value).lower() in {"0", "false", "no", "off"}:
+            parsed = False
+        else:
+            raise ValueError("expected a boolean")
+    elif schema.type == "string" or schema.type is None:
+        parsed = str(value)
+    else:
+        parsed = value
+    if schema.enum and parsed not in schema.enum:
+        raise ValueError("value is not in the allowed enumeration")
+    if isinstance(parsed, int | float):
+        if schema.minimum is not None and parsed < schema.minimum:
+            raise ValueError(f"value must be at least {schema.minimum}")
+        if schema.maximum is not None and parsed > schema.maximum:
+            raise ValueError(f"value must be at most {schema.maximum}")
+    if isinstance(parsed, str):
+        if schema.min_length is not None and len(parsed) < schema.min_length:
+            raise ValueError(f"value is shorter than {schema.min_length}")
+        if schema.max_length is not None and len(parsed) > schema.max_length:
+            raise ValueError(f"value is longer than {schema.max_length}")
+    return parsed
+
+
 def _schema_default(schema: Schema) -> Any:
     if schema.default is not None:
         return schema.default

app/main.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from fastapi import FastAPI
 
-from app.api.errors import unhandled_exception_handler
+from app.api.errors import ApiError, api_error_handler, unhandled_exception_handler
 from app.api.middleware import RequestContextMiddleware
 from app.api.registry import HandlerRegistry, register_contract_routes
 from app.config import Settings, get_settings
@@ -30,6 +30,7 @@ def create_app(
     )
     app.add_middleware(RequestContextMiddleware, header_name=resolved.request_id_header)
     app.add_exception_handler(Exception, unhandled_exception_handler)
+    app.add_exception_handler(ApiError, api_error_handler)
     app.include_router(health_router)
     if resolved.contract_snapshot is not None:
         snapshot = Snapshot.model_validate_json(resolved.contract_snapshot.read_bytes())

tests/unit/test_compatible_io.py

@@ -0,0 +1,87 @@
+"""Golden HTTP input/output compatibility checks."""
+
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from fastapi import Request
+from httpx import ASGITransport, AsyncClient
+
+from app.api.registry import HandlerRegistry
+from app.config import Settings
+from app.contracts.model import Method, Parameter, PathContract, Schema, Snapshot
+from app.main import create_app
+from tests.unit.test_health import FakeDatabase
+
+
+async def client_for(tmp_path: Path) -> AsyncClient:
+    method = Method(
+        verb="POST",
+        name="update",
+        parameters=(
+            Parameter(name="node", definition=Schema(type="string")),
+            Parameter(name="count", definition=Schema(type="integer", minimum=1)),
+            Parameter(name="force", definition=Schema(type="boolean", optional=True)),
+        ),
+        returns=Schema(type="null"),
+        checksum="1" * 64,
+    )
+    snapshot = Snapshot(
+        source_version="test",
+        retrieved_at=datetime(2026, 1, 1, tzinfo=UTC),
+        raw_sha256="0" * 64,
+        paths=(PathContract(path="/nodes/{node}/test", methods=(method,)),),
+        path_count=1,
+        method_count=1,
+    )
+    path = tmp_path / "snapshot.json"
+    path.write_bytes(snapshot.canonical_bytes())
+    handlers = HandlerRegistry()
+
+    async def handler(_request: Request, inputs: dict[str, Any]) -> None:
+        assert inputs["values"]["count"] >= 1
+        return None
+
+    handlers.register("/nodes/{node}/test", "POST", handler)
+    app = create_app(
+        Settings(contract_snapshot=path), lambda _settings: FakeDatabase(True), handlers
+    )
+    return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
+
+
+async def test_json_input_and_null_envelope(tmp_path: Path) -> None:
+    async with await client_for(tmp_path) as client:
+        response = await client.post("/api2/json/nodes/pve/test", json={"count": 2, "force": True})
+
+    assert response.status_code == 200
+    assert response.json() == {"data": None}
+
+
+async def test_form_input_and_validation_error_shape(tmp_path: Path) -> None:
+    async with await client_for(tmp_path) as client:
+        valid = await client.post(
+            "/api2/json/nodes/pve/test",
+            content="count=1&force=yes",
+            headers={"content-type": "application/x-www-form-urlencoded"},
+        )
+        invalid = await client.post("/api2/json/nodes/pve/test", json={"count": 0, "unknown": "x"})
+
+    assert valid.status_code == 200
+    assert invalid.status_code == 400
+    assert invalid.json() == {
+        "data": None,
+        "message": "parameter verification failed",
+        "errors": {
+            "count": "value must be at least 1",
+            "unknown": "property is not defined in schema",
+        },
+    }
+
+
+async def test_non_object_json_is_rejected_without_fastapi_body(tmp_path: Path) -> None:
+    async with await client_for(tmp_path) as client:
+        response = await client.post("/api2/json/nodes/pve/test", json=[1, 2])
+
+    assert response.status_code == 400
+    assert response.json()["errors"] == {"body": "expected an object"}
+    assert "detail" not in response.json()