Test API servers with POST requests, query parameters, a JSON body and an X-API-Key header

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 - `datacontract breaking` now allows to select a `--perspective` (either `consumer` or `producer`)
+- `datacontract test`: API servers send POST requests with query parameters and a JSON body, set by the `method`, `params` and `body` custom properties, and an `X-API-Key` header from `DATACONTRACT_API_HEADER_X_API_KEY` (also used by `datacontract import odata`) (#1747)
+- `datacontract import openapi` imports POST operations, and writes required query parameters to the `params` custom property (#1747)
+
+### Security
+- `datacontract api`: a posted contract's `api` server may not reach private network addresses unless the server runs with `--allow-private-network`, and is refused while `DATACONTRACT_API_HEADER_AUTHORIZATION` or `DATACONTRACT_API_HEADER_X_API_KEY` is set on the server (#1747)
 
 ### Fixed
 - `datacontract breaking` grades from the consumer's perspective by default (previously inconsistent mix of consumer and producer perspective)

datacontract/api.py

@@ -259,9 +259,10 @@ def _cli_version() -> str:
 untrusted input: a `quality.type: sql` rule must be a read-only query, a credential held in the
 server's environment is never sent to a host the contract names, a `publish_url` may only point at
 the Entropy Data platform or the host configured via `ENTROPY_DATA_HOST`, and `servers[].type: local`
-is refused so a caller cannot read the server's own files. Set
-`DATACONTRACT_CLI_API_ALLOW_LOCAL_FILES=true` if the deployment serves its own files on purpose;
-the contract is then still confined to the paths it declares.
+is refused so a caller cannot read the server's own files, and an `api` server's requests may not
+reach a private network address. Set `DATACONTRACT_CLI_API_ALLOW_LOCAL_FILES=true` if the deployment
+serves its own files on purpose; the contract is then still confined to the paths it declares. Set
+`DATACONTRACT_CLI_API_ALLOW_PRIVATE_NETWORK=true` to test APIs in the server's own network.
 
 ## Connecting to data sources
 
@@ -525,6 +526,8 @@ def _parse_filters_query(filters: str | None) -> dict[str, str] | None:
     # option pins it), so an environment-held SASL password can never be paired
     # with a posted contract safely -- host_field is None, so it is always rejected.
     "kafka": ("broker host", None, ("kafka_sasl_password",)),
+    # Nothing pins an API server's location either.
+    "api": ("location", None, ("api_header_authorization", "api_header_x_api_key")),
 }
 
 
@@ -552,6 +555,7 @@ def _selected_server(body: str, server_name: str | None, config):
 _LOCAL_SERVER_TYPES = frozenset({"local", "duckdb"})
 
 ALLOW_LOCAL_FILES_ENV = "DATACONTRACT_CLI_API_ALLOW_LOCAL_FILES"
+ALLOW_PRIVATE_NETWORK_ENV = "DATACONTRACT_CLI_API_ALLOW_PRIVATE_NETWORK"
 _TRUTHY = frozenset({"1", "true", "yes", "on"})
 
 
@@ -570,6 +574,11 @@ def _local_files_allowed() -> bool:
     return (os.getenv(ALLOW_LOCAL_FILES_ENV) or "").strip().lower() in _TRUTHY
 
 
+def _private_network_allowed() -> bool:
+    """Whether the operator lets a posted contract's API requests reach private network addresses."""
+    return (os.getenv(ALLOW_PRIVATE_NETWORK_ENV) or "").strip().lower() in _TRUTHY
+
+
 def _reject_local_server_type(body: str, server_name: str | None, config) -> None:
     """Keep a posted contract from reading the server's own disk.
 
@@ -623,20 +632,23 @@ def _reject_environment_credentials_to_untrusted_host(body: str, server_name: st
         return  # no environment-held secret for this server type; nothing to protect
     if host_field is not None and resolved.option_source(host_field) == "env":
         return  # the operator pinned the target host; the contract's host is ignored
+    held = [
+        env_name(field, Config.model_fields[field]) for field in secret_fields if resolved.option_source(field) == "env"
+    ]
     if host_field is not None:
         env_var = env_name(host_field, Config.model_fields[host_field])
         detail = (
-            f"This server holds a {server.type} credential in its environment, so the connection "
-            f"{target_attr} must also come from the server's environment ({env_var}); it will not be "
-            f"taken from the posted data contract. Set {env_var}, or send the full set of credentials "
-            f"with the request instead of relying on the server's environment."
+            f"This server holds a credential for {server.type} servers in its environment ({', '.join(held)}), so the "
+            f"connection {target_attr} must also come from the server's environment ({env_var}); it will not "
+            f"be taken from the posted data contract. Set {env_var}, or send the full set of credentials with "
+            f"the request instead of relying on the server's environment."
         )
     else:
         detail = (
-            f"This server holds a {server.type} credential in its environment, but its {target_attr} "
-            f"can only come from the posted data contract, so the credential cannot be pinned to a "
-            f"trusted target. Send the credentials with the request instead of relying on the server's "
-            f"environment."
+            f"This server holds a credential for {server.type} servers in its environment ({', '.join(held)}), but "
+            f"their {target_attr} can only come from the posted data contract, so the credential cannot be pinned to "
+            f"a trusted target. Send the credentials as request headers instead "
+            f"({', '.join(name.lower().replace('_', '-') for name in held)})."
         )
     raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=detail)
 
@@ -835,6 +847,7 @@ def test(
         filters=parsed_filters,
         untrusted_contract=untrusted_contract,
         allowed_variables=_allowed_contract_variables(),
+        allow_private_network=_private_network_allowed(),
     ).test()
 
 

datacontract/command_api.py

@@ -57,6 +57,14 @@ def api(
             help="Let a posted data contract read the server's own disk through servers[].type: local. Off by default.",
         ),
     ] = None,
+    allow_private_network: Annotated[
+        bool | None,
+        typer.Option(
+            "--allow-private-network/--no-allow-private-network",
+            help="Let a posted data contract's API requests (servers[].type: api) reach private network addresses, "
+            "such as localhost or the cloud metadata endpoint. Off by default.",
+        ),
+    ] = None,
     debug: debug_option = None,
 ):
     """
@@ -78,12 +86,14 @@ def api(
     enable_debug_logging(debug)
 
     # Passed through the environment, which survives uvicorn's --reload respawn.
-    from datacontract.api import ALLOW_LOCAL_FILES_ENV
+    from datacontract.api import ALLOW_LOCAL_FILES_ENV, ALLOW_PRIVATE_NETWORK_ENV
 
     if contract_variables is not None:
         os.environ[CONTRACT_VARIABLES_ENV] = contract_variables
     if allow_local_files is not None:
         os.environ[ALLOW_LOCAL_FILES_ENV] = "true" if allow_local_files else "false"
+    if allow_private_network is not None:
+        os.environ[ALLOW_PRIVATE_NETWORK_ENV] = "true" if allow_private_network else "false"
 
     import uvicorn
     from uvicorn.config import LOGGING_CONFIG

datacontract/command_import.py

@@ -1053,16 +1053,17 @@ def import_openapi(
     operation: Annotated[
         Optional[str],
         typer.Option(
-            help="The GET operation to import, by operationId or path (e.g. listOrders or /orders). "
-            "Required when the document has more than one GET operation."
+            help="The GET or POST operation to import, by operationId, method and path, or a GET operation's path "
+            "(e.g. listOrders, 'POST /orders/search', or /orders). Required unless the document has exactly one GET "
+            "operation."
         ),
     ] = None,
     output: output_option = None,
     owner: owner_option = None,
     id: id_option = None,
     debug: debug_option = None,
 ):
-    """Import a data contract from the response of a GET operation in an OpenAPI 3.x document."""
+    """Import a data contract from the response of a GET or POST operation in an OpenAPI 3.x document."""
     enable_debug_logging(debug)
     result = DataContract.import_from_source(
         config=cli_config(), format="openapi", source=source, openapi_operation=operation, owner=owner, id=id

datacontract/config/settings.py

@@ -113,6 +113,7 @@ class Config(BaseSettings):
 
     # general
     api_header_authorization: SecretStr | None = None
+    api_header_x_api_key: SecretStr | None = None
     custom_quality_checks: str | None = None
     max_errors: int | None = None
     connect_timeout: int | None = None
@@ -526,6 +527,9 @@ def get_datacontract_manager_host(self, required: bool = False) -> str | None:
     def get_api_header_authorization(self, required: bool = False) -> str | None:
         return self._str_option("api_header_authorization", required)
 
+    def get_api_header_x_api_key(self, required: bool = False) -> str | None:
+        return self._str_option("api_header_x_api_key", required)
+
     # --- custom quality checks ---
     def get_custom_quality_checks(self) -> str | None:
         return self._str_option("custom_quality_checks")

datacontract/config/variables.py

@@ -175,7 +175,8 @@ def resolve_server_variables(server: Server, variables: Optional[Mapping[str, st
 
     ``port`` may hold a reference (the schema allows a string there for that
     reason); a resolved all-digit port becomes an int again. Custom property
-    values are resolved too. The contract's server is not mutated.
+    values are resolved too, nested ones like an API request body included.
+    The contract's server is not mutated.
     """
     updates = {}
     for field in Server.model_fields:
@@ -192,17 +193,15 @@ def resolve_server_variables(server: Server, variables: Optional[Mapping[str, st
         custom_properties = []
         changed = False
         for prop in server.customProperties:
+            source = f"server '{server.server}' custom property '{prop.property}'"
             if contains_variables(prop.value):
                 changed = True
                 prop = prop.model_copy(
-                    update={
-                        "value": resolve_variables(
-                            prop.value,
-                            source=f"server '{server.server}' custom property '{prop.property}'",
-                            variables=variables,
-                        )
-                    }
+                    update={"value": resolve_variables(prop.value, source=source, variables=variables)}
                 )
+            elif isinstance(prop.value, (dict, list)):
+                changed = True
+                prop = prop.model_copy(update={"value": resolve_runtime_variables(prop.value, source, variables)})
             custom_properties.append(prop)
         if changed:
             updates["customProperties"] = custom_properties

datacontract/data_contract.py

@@ -114,6 +114,7 @@ def __init__(
         dry_run: bool = False,
         untrusted_contract: bool = False,
         allowed_variables: list[str] | None = None,
+        allow_private_network: bool = False,
         config: "Config | dict[str, str] | None" = None,
     ):
         self._data_contract_file = data_contract_file
@@ -145,6 +146,8 @@ def __init__(
         self._untrusted_contract = untrusted_contract
         # fnmatch globs of the environment variables an untrusted contract may read.
         self._allowed_variables = allowed_variables
+        # Whether an untrusted contract's API requests may reach private network addresses.
+        self._allow_private_network = allow_private_network
         self._config = Config.resolve(config)
 
     @classmethod
@@ -259,6 +262,7 @@ def test(self) -> Run:
                 config=self._config,
                 untrusted_contract=self._untrusted_contract,
                 allowed_variables=self._allowed_variables,
+                allow_private_network=self._allow_private_network,
             )
 
         except DataContractException as e:

datacontract/engines/data_contract_test.py

@@ -3,6 +3,7 @@
 import tempfile
 import typing
 from typing import Mapping
+from urllib.parse import parse_qsl, urlsplit
 
 import requests
 import yaml
@@ -28,6 +29,7 @@
     check_that_datacontract_contains_valid_server_configuration,
 )
 from datacontract.engines.fastjsonschema.check_jsonschema import check_jsonschema
+from datacontract.engines.http_api import ApiSession, PrivateAddressRefused, PublicAddressAdapter, api_headers
 from datacontract.engines.ibis.ibis_check_execute import build_check_stubs, execute_ibis_checks, set_result
 from datacontract.lint.resolve import _SafeLoaderNoTimestamp
 from datacontract.model.exceptions import DataContractException
@@ -54,6 +56,7 @@ def execute_data_contract_test(
     config: Config | None = None,
     untrusted_contract: bool = False,
     allowed_variables: list[str] | None = None,
+    allow_private_network: bool = False,
 ):
     config = Config.resolve(config)
     # An untrusted contract reads only the allow-listed variables.
@@ -109,18 +112,13 @@ def execute_data_contract_test(
         check_that_quality_ids_exist(data_contract, quality_ids, schema_name)
 
     if server.type == "api":
-        if dry_run:
-            # A dry run reads nothing, so the response that would become the
-            # local file is never fetched; stand in the server it would return.
-            server = Server(
-                server="api_local",
-                type="local",
-                format="json",
-                path="(not fetched: dry run)",
-                delimiter=server.delimiter,
-            )
-        else:
-            server = process_api_response(run, server, config)
+        server = process_api_response(
+            run,
+            server,
+            config,
+            dry_run=dry_run,
+            public_addresses_only=untrusted_contract and not allow_private_network,
+        )
 
     model_filters = resolve_row_filters(data_contract, server, run, filter, filters, schema_name)
 
@@ -436,22 +434,79 @@ def _report_dry_run(
         run.log_warn("dry run: file-metadata checks for blob schemas are not included in the plan")
 
 
-def process_api_response(run, server, config: Config | None = None):
+def process_api_response(
+    run, server, config: Config | None = None, dry_run: bool = False, public_addresses_only: bool = False
+):
     config = Config.resolve(config)
+    custom_properties = {prop.property: prop.value for prop in server.customProperties or []}
+    method = str(custom_properties.get("method", "GET")).upper()
+    if method not in ("GET", "POST"):
+        raise _api_request_error(f"Method {custom_properties['method']!r} is not supported; use GET or POST.")
+    if "body" in custom_properties and method != "POST":
+        raise _api_request_error("A body requires method POST.")
+    if not isinstance(custom_properties.get("params", {}), dict):
+        raise _api_request_error("The params custom property must map each query parameter name to its value.")
+    params = {}
+    for name, value in custom_properties.get("params", {}).items():
+        values = value if isinstance(value, list) else [value]
+        if not all(isinstance(v, (str, int, float, bool)) for v in values):
+            raise _api_request_error(
+                f"Query parameter {name!r} must be a string, number, or boolean, or a list of them."
+            )
+        params[name] = [str(v).lower() if isinstance(v, bool) else v for v in values]
+    clashes = sorted(
+        set(params) & {name for name, _ in parse_qsl(urlsplit(server.location).query, keep_blank_values=True)}
+    )
+    if clashes:
+        raise _api_request_error(
+            f"Query parameter {clashes[0]!r} is set both in the location and in the params custom property."
+        )
+
+    if dry_run:
+        # A dry run reads nothing, so the response that would become the
+        # local file is never fetched; stand in the server it would return.
+        return Server(
+            server="api_local",
+            type="local",
+            format="json",
+            path="(not fetched: dry run)",
+            delimiter=server.delimiter,
+        )
+
     tmp_dir = tempfile.TemporaryDirectory(prefix="datacontract_cli_api_")
     atexit.register(tmp_dir.cleanup)
-    headers = {"Accept": "application/json, application/yaml;q=0.9, */*;q=0.8"}
-    if config.get_api_header_authorization() is not None:
-        headers["Authorization"] = config.get_api_header_authorization()
+    headers = {"Accept": "application/json, application/yaml;q=0.9, */*;q=0.8", **api_headers(config)}
+    body = custom_properties.get("body")
+    # a string is the JSON text as written, anything else is converted to JSON
+    if isinstance(body, str):
+        headers["Content-Type"] = "application/json"
     try:
-        response = requests.get(server.location, headers=headers)
-        response.raise_for_status()
+        with ApiSession() as session:
+            if public_addresses_only:
+                session.trust_env = False
+                session.mount("http://", PublicAddressAdapter())
+                session.mount("https://", PublicAddressAdapter())
+            response = session.request(
+                method,
+                server.location,
+                params=params,
+                headers=headers,
+                timeout=(config.get_connect_timeout(), None),
+                data=body.encode() if isinstance(body, str) else None,
+                json=None if isinstance(body, str) else body,
+            )
+            response.raise_for_status()
     except requests.exceptions.RequestException as e:
+        # urllib3 wraps the refusal in its retry error
+        refusal = getattr(e.args[0], "reason", None) if e.args else None
+        error = refusal.message if isinstance(refusal, PrivateAddressRefused) else e
+        # the API's own explanation, such as which part of the body it rejected
+        answer = f" {e.response.text[:500]}" if e.response is not None and e.response.text else ""
         raise DataContractException(
             type="connection",
             name="API server connection error",
             result=ResultEnum.error,
-            reason=f"Failed to fetch API response from {server.location}: {e}",
+            reason=f"Failed to fetch API response from {server.location}: {error}{answer}",
             engine="datacontract-cli",
         )
     content = response.text
@@ -482,6 +537,16 @@ def process_api_response(run, server, config: Config | None = None):
     return new_server
 
 
+def _api_request_error(reason: str) -> DataContractException:
+    return DataContractException(
+        type="general",
+        name="Build the API request",
+        result=ResultEnum.failed,
+        reason=reason,
+        engine="datacontract-cli",
+    )
+
+
 def _has_blob_schemas(data_contract: OpenDataContractStandard, schema_name: str) -> bool:
     """Return True if the (possibly filtered) schema list contains any logicalType='blob' schema."""
     if data_contract.schema_ is None: