| """HTTP client for tests, with virtual-host resolution. |
| |
| Port of the request side of Apache::TestRequest. Wraps an ``httpx.Client`` and |
| exposes the request helpers translated tests use most: GET/POST/HEAD/OPTIONS/PUT |
| plus the ``*_BODY`` / ``*_RC`` shorthands, vhost selection (the analog of |
| ``Apache::TestRequest::module``), and runtime introspection gates |
| (``have_min_apache_version``, ``have_module``, ``vars``). |
| |
| By default requests are NOT redirected (matching how the Perl tests assert on |
| 3xx explicitly); pass ``redirect_ok=True`` to follow them. |
| """ |
| |
| from __future__ import annotations |
| |
| import ssl |
| from pathlib import Path |
| |
| import httpx |
| |
| from .config import TestConfig |
| |
| |
| class TestClient: |
| def __init__(self, config: TestConfig, *, timeout: float = 30.0) -> None: |
| self.config = config |
| self._timeout = timeout |
| self._scheme = "http" |
| self._module: str | None = None # selected vhost (Apache::TestRequest::module) |
| # httpx binds verify/cert at client construction, so we keep a small |
| # cache of clients keyed by the client-cert name (None = no cert). All |
| # verify against the generated test CA (server uses self-signed certs). |
| self._clients: dict[str | None, httpx.Client] = {} |
| self._default_client = httpx.Client(timeout=timeout, follow_redirects=False) |
| |
| # -- SSL: test-CA verification + client certs (Apache::TestRequest cert=>) -- |
| |
| def _ca_path(self) -> Path | None: |
| """Path to the generated test CA cert, or None if no SSL CA was built.""" |
| ca = Path(self.config.vars["sslca"]) / "asf" / "certs" / "ca.crt" |
| return ca if ca.exists() else None |
| |
| def _ssl_context(self, *, client_cert: str | None = None) -> ssl.SSLContext | bool: |
| """Build an SSL context trusting the test CA (and optionally a client cert). |
| |
| Returns a configured ``ssl.SSLContext``, or ``True`` (httpx default |
| verification) if no test CA was generated. Using a context avoids httpx's |
| deprecated ``verify=<str>`` path. |
| """ |
| ca = self._ca_path() |
| if ca is None: |
| return True |
| ctx = ssl.create_default_context(cafile=str(ca)) |
| if client_cert is not None: |
| # Per-directory `SSLVerifyClient require` requests the client cert |
| # mid-connection. Under TLS 1.3 that is Post-Handshake Authentication, |
| # which must be enabled on the client context BEFORE loading the cert |
| # chain (the flag is read at handshake setup). With it on, httpx/the |
| # CPython ssl client perform PHA correctly and the server returns 200 |
| # -- so we keep TLS 1.3 (more faithful than capping to 1.2) and let |
| # PHA-specific tests (pha.t, CVE-2019-0215) exercise the real path. |
| ctx.post_handshake_auth = True |
| ctx.load_cert_chain(certfile=self._client_cert(client_cert)) |
| return ctx |
| |
| def _client_cert(self, name: str) -> str: |
| """Resolve a client-cert name (e.g. 'client_ok') to its combined PEM. |
| |
| Apache::TestRequest's ``cert => 'name'`` uses the proxy/<name>.pem file |
| (cert + key concatenated) generated by the SSL CA. httpx accepts that |
| single combined PEM as ``cert=``. |
| """ |
| pem = Path(self.config.vars["sslca"]) / "asf" / "proxy" / f"{name}.pem" |
| if not pem.exists(): |
| raise FileNotFoundError(f"no client cert PEM for {name!r}: {pem}") |
| return str(pem) |
| |
| def _client_for(self, cert: str | None) -> httpx.Client: |
| """Return (memoized) an httpx client for the given client-cert name.""" |
| if cert is None: |
| return self._default_client |
| if cert not in self._clients: |
| self._clients[cert] = httpx.Client( |
| timeout=self._timeout, |
| follow_redirects=False, |
| verify=self._ssl_context(client_cert=cert), |
| ) |
| return self._clients[cert] |
| |
| @property |
| def _client(self) -> httpx.Client: |
| """The HTTPS-verifying client when scheme is https, else the plain one. |
| |
| For https without an explicit client cert we still need to trust the |
| test CA, so route through a cert=None client that sets verify=CA. |
| """ |
| if self._scheme == "https": |
| key = "__https_noclientcert__" |
| if key not in self._clients: |
| self._clients[key] = httpx.Client( |
| timeout=self._timeout, |
| follow_redirects=False, |
| verify=self._ssl_context(), |
| ) |
| return self._clients[key] |
| return self._default_client |
| |
| # -- introspection (Apache::Test::vars / have_*) ---------------------- |
| |
| @property |
| def servername(self) -> str: |
| return self.config.vars["servername"] |
| |
| def vars(self, key: str | None = None): |
| """Apache::Test::vars() -- the whole vars dict, or one key.""" |
| return self.config.vars if key is None else self.config.vars.get(key) |
| |
| def have_module(self, name: str) -> bool: |
| """have_module('x') -- True if the module is loaded in the server.""" |
| return self.config.info.has_module(name) |
| |
| def have_min_apache_version(self, version: str) -> bool: |
| """have_min_apache_version('2.4.x') -- runtime version gate.""" |
| parts = tuple(int(x) for x in str(version).split(".")) |
| parts += (0,) * (3 - len(parts)) |
| return self.config.info.version >= parts |
| |
| def have_apache(self, major: int) -> bool: |
| """have_apache(2) -- True if the server major version equals ``major``.""" |
| return self.config.info.version[0] == major |
| |
| def apxs(self, query: str) -> str | None: |
| """Query apxs (apxs -q VAR), e.g. apxs("INCLUDEDIR"). None if no apxs. |
| |
| The Python analog of Apache::TestConfig->apxs(...); used by tests like |
| mmn.t that read installed httpd headers. |
| """ |
| if self.config.apxs is None: |
| return None |
| import subprocess |
| |
| proc = subprocess.run( # noqa: S603 - trusted apxs path |
| ["perl", str(self.config.apxs), "-q", query], |
| capture_output=True, |
| text=True, |
| check=False, |
| ) |
| return proc.stdout.strip() if proc.returncode == 0 else None |
| |
| # -- request configuration (Apache::TestRequest::scheme/module) ------- |
| |
| def scheme(self, scheme: str) -> None: |
| self._scheme = scheme |
| |
| def module(self, name: str | None) -> None: |
| """Select a virtual host by module name for subsequent requests.""" |
| self._module = name |
| |
| # -- URL construction ------------------------------------------------- |
| |
| def _base_port(self) -> int: |
| """The default port for the current scheme (no module selected). |
| |
| https uses the mod_ssl vhost's port if present; otherwise the main port. |
| """ |
| if self._scheme == "https": |
| ssl_name = self.config.vars.get("ssl_module_name", "mod_ssl") |
| vhost = self.config.vhosts.get(ssl_name) |
| if vhost is not None: |
| return vhost.port |
| return int(self.config.vars["port"]) |
| |
| def _port(self) -> int: |
| """Port for the currently-selected module (lenient: unknown -> base port).""" |
| if self._module is not None: |
| vhost = self.config.vhosts.get(self._module) |
| if vhost is not None: |
| return vhost.port |
| return self._base_port() |
| |
| @property |
| def base_url(self) -> str: |
| return f"{self._scheme}://{self.servername}:{self._port()}" |
| |
| def vhost_port(self, module: str) -> int: |
| """Resolve a *configured* vhost's port; raise if the module has no vhost. |
| |
| Strict lookup used by vhost_url(), where a missing vhost almost always |
| means a test typo. For the lenient fallback behaviour (unknown module -> |
| main server port) that Apache::TestRequest::hostport/vhost_socket use, |
| see :meth:`resolve_port`. |
| """ |
| vhost = self.config.vhosts.get(module) |
| if vhost is None: |
| raise KeyError( |
| f"no virtual host configured for module {module!r}; " |
| f"known: {sorted(self.config.vhosts)}" |
| ) |
| return vhost.port |
| |
| def resolve_port(self, module: str | None) -> int: |
| """Port for ``module``, falling back to the main port if it has no vhost. |
| |
| Mirrors Apache::TestRequest::hostport: ``$vhosts{$module}{hostport}`` |
| with a fall-through to the default ``servername:port`` when the module |
| isn't a configured vhost (e.g. the "h2c" pseudo-module in |
| CVE-2017-7659, which just exercises the main server). ``"default"`` and |
| ``None`` both mean the main port. |
| """ |
| if module is None or module == "default": |
| return self._base_port() |
| vhost = self.config.vhosts.get(module) |
| return vhost.port if vhost is not None else self._base_port() |
| |
| def vhost_url(self, module: str, path: str = "/") -> str: |
| if not path.startswith("/"): |
| path = "/" + path |
| return f"{self._scheme}://{self.servername}:{self.vhost_port(module)}{path}" |
| |
| def hostport(self, module: str | None = None) -> str: |
| """host:port for the selected/given vhost (Apache::TestRequest::hostport). |
| |
| Unknown module names fall back to the main server port, matching Perl. |
| """ |
| module = module if module is not None else self._module |
| return f"{self.servername}:{self.resolve_port(module)}" |
| |
| # -- raw sockets (Apache::TestRequest::vhost_socket / getline) -------- |
| |
| def vhost_socket(self, module: str | None = None, *, timeout: float = 10.0): |
| """Open a raw socket to a vhost (defaults to the selected/main vhost). |
| |
| Returns a :class:`apache_pytest.rawsocket.VhostSocket` for protocol-level |
| tests that send hand-built requests and read raw response lines. |
| """ |
| from .rawsocket import open_vhost_socket |
| |
| module = module if module is not None else self._module |
| # Lenient resolution (unknown module -> main port), like Perl's |
| # vhost_socket; covers pseudo-modules such as "h2c" that have no vhost. |
| port = self.resolve_port(module) |
| use_ssl = self._scheme == "https" or (module is not None and "ssl" in module) |
| return open_vhost_socket( |
| self.servername, port, use_ssl=use_ssl, timeout=timeout |
| ) |
| |
| def _url(self, path: str) -> str: |
| """Absolute URLs pass through; bare paths get the current base_url.""" |
| if path.startswith(("http://", "https://")): |
| return path |
| if not path.startswith("/"): |
| path = "/" + path |
| return self.base_url + path |
| |
| # -- requests (Apache::TestRequest GET/POST/HEAD/...) ----------------- |
| |
| def _request( |
| self, |
| method: str, |
| path: str, |
| *, |
| redirect_ok: bool = False, |
| cert: str | None = None, |
| **kwargs: object, |
| ) -> httpx.Response: |
| # cert => 'name' selects a client-cert-bearing client (Apache::TestRequest |
| # cert=> option); cert=None over https still verifies against the test CA. |
| client = self._client_for(cert) if cert is not None else self._client |
| return client.request( |
| method, self._url(path), follow_redirects=redirect_ok, **kwargs # type: ignore[arg-type] |
| ) |
| |
| def GET(self, path: str, **kwargs: object) -> httpx.Response: |
| return self._request("GET", path, **kwargs) |
| |
| def raw_response( |
| self, |
| method: str, |
| path: str, |
| *, |
| cert: str | None = None, |
| **kwargs: object, |
| ) -> httpx.Response: |
| """Like :meth:`_request` but WITHOUT content-decoding the body. |
| |
| httpx transparently inflates gzip/deflate responses, so ``.content`` is |
| the decoded plaintext (while ``Content-Encoding`` is left in place). |
| Tests that need the bytes exactly as they came off the wire -- e.g. the |
| mod_deflate round-trips that re-POST the gzip through an inflate filter, |
| or mod_reflector asserting the body was actually transformed -- can't use |
| that. Stream the response, read ``iter_raw()`` (the undecoded bytes, the |
| analog of LWP not auto-decoding), and stash them on ``.raw_content`` so |
| callers still see ``.status_code`` and ``.headers``. |
| """ |
| client = self._client_for(cert) if cert is not None else self._client |
| request = client.build_request(method, self._url(path), **kwargs) # type: ignore[arg-type] |
| response = client.send(request, stream=True) |
| try: |
| response.raw_content = b"".join(response.iter_raw()) # type: ignore[attr-defined] |
| finally: |
| response.close() |
| return response |
| |
| def GET_RAW(self, path: str, **kwargs: object) -> bytes: |
| """GET ``path`` and return the raw, undecoded response body bytes.""" |
| return self.raw_response("GET", path, **kwargs).raw_content # type: ignore[attr-defined] |
| |
| def HEAD(self, path: str, **kwargs: object) -> httpx.Response: |
| return self._request("HEAD", path, **kwargs) |
| |
| def OPTIONS(self, path: str, **kwargs: object) -> httpx.Response: |
| return self._request("OPTIONS", path, **kwargs) |
| |
| def PUT(self, path: str, **kwargs: object) -> httpx.Response: |
| return self._request("PUT", path, **kwargs) |
| |
| def POST(self, path: str, content: object = None, **kwargs: object) -> httpx.Response: |
| if content is not None and "content" not in kwargs and "data" not in kwargs: |
| kwargs["content"] = content |
| return self._request("POST", path, **kwargs) |
| |
| # body / status-code shorthands (GET_BODY, GET_RC, POST_BODY, ...) |
| def GET_BODY(self, path: str, **kwargs: object) -> str: |
| return self.GET(path, **kwargs).text |
| |
| def GET_RC(self, path: str, **kwargs: object) -> int: |
| # Apache::TestRequest::GET_RC returns the response code; LWP surfaces a |
| # transport/TLS failure (e.g. the server aborting with a "certificate |
| # revoked" alert) as a 5xx response rather than throwing. Mirror that so |
| # access-control tests that assert ``!= 200`` see a non-200 code. |
| try: |
| return self.GET(path, **kwargs).status_code |
| except httpx.TransportError: |
| return 500 |
| |
| def POST_BODY(self, path: str, content: object = None, **kwargs: object) -> str: |
| return self.POST(path, content=content, **kwargs).text |
| |
| def HEAD_RC(self, path: str, **kwargs: object) -> int: |
| try: |
| return self.HEAD(path, **kwargs).status_code |
| except httpx.TransportError: |
| return 500 |
| |
| # lowercase aliases for ergonomic pytest-style calls |
| def get(self, path: str, **kwargs: object) -> httpx.Response: |
| return self.GET(path, **kwargs) |
| |
| def post(self, path: str, **kwargs: object) -> httpx.Response: |
| return self.POST(path, **kwargs) |
| |
| def request(self, method: str, url: str, **kwargs: object) -> httpx.Response: |
| return self._client.request(method, url, **kwargs) # type: ignore[arg-type] |
| |
| def close(self) -> None: |
| self._default_client.close() |
| for client in self._clients.values(): |
| client.close() |
| self._clients.clear() |
