diff --git a/Documentation/authentication-and-tls.md b/Documentation/authentication-and-tls.md new file mode 100644 index 0000000..513e1a9 --- /dev/null +++ b/Documentation/authentication-and-tls.md @@ -0,0 +1,38 @@ +--- +title: Authentication and TLS +description: How the Chronicle Python client obtains and refreshes OAuth tokens, which certificates it trusts, and what happens when authentication fails. +--- + +## Token handling + +`OAuthTokenProvider` posts `grant_type=client_credentials`, `client_id` and `client_secret` as +`application/x-www-form-urlencoded` to `/connect/token` on the kernel's own host and port. The channel asks the +provider for the current token when each new gRPC call starts and sends it as `authorization: Bearer `; no +token is baked into the channel. + +- A token is reused until shortly before it expires: half of its lifetime for short tokens, otherwise 30 seconds + early. +- When the response has no `expires_in`, the token is trusted for 30 seconds only. +- Concurrent callers share one request. Cancelling one caller does not cancel the request the others wait for. +- A response that is not JSON, has no `access_token`, a `token_type` other than `Bearer`, or an invalid `expires_in` + raises `TokenResponseError`. HTTP 400, 401 and 403 raise `TokenAuthorizationError` with the OAuth `error` and + `error_description`. Redirects are never followed, so the secret only goes to the configured host. +- The client secret and tokens are excluded from `repr()` and from every error message, including text the server + echoes back. +- A call that receives `UNAUTHENTICATED` is not retried. Reconnect and retry behavior is not part of this milestone. + +## Certificate trust + +Certificates and host names are verified for every connection. `skipTlsValidation` is `true` by default in the +connection string, as in the .NET client; this client honors it with these rules: + +| Situation | Trusted certificate | +| --- | --- | +| `ca_certificates=` is passed to `ChronicleClient.connect` | Exactly those roots | +| `skipTlsValidation` is `true` and the host is `localhost` or a loopback address | The certificate the local kernel presents, read once at connect time; the host name is still checked | +| `skipTlsValidation=false` | The platform's default roots | +| `skipTlsValidation` is `true` and the host is **not** loopback | The platform's default roots. The option is ignored | + +A remote kernel is therefore never reached with relaxed validation. To reach a remote kernel with a private +certificate authority, pass its certificate as `ca_certificates`. A loopback kernel with a self-signed certificate +works with the defaults; add `?skipTlsValidation=false` to prove it fails without trust. diff --git a/Documentation/client-development-guide.md b/Documentation/client-development-guide.md index d55162a..5da1c16 100644 --- a/Documentation/client-development-guide.md +++ b/Documentation/client-development-guide.md @@ -22,15 +22,15 @@ Chronicle C# contracts ### Temporary contracts distribution `cratis-chronicle-contracts` is not published to PyPI. The project dependency resolves the verified -`cratis-chronicle-contracts` 16.38.2 wheel from its matching -[Chronicle GitHub release](https://github.com/Cratis/Chronicle/releases/tag/v16.38.2), pinned by SHA-256 in +`cratis-chronicle-contracts` 19.31.3 wheel from its matching +[Chronicle GitHub release](https://github.com/Cratis/Chronicle/releases/tag/v19.31.3), pinned by SHA-256 in `pyproject.toml`. A normal development install fetches it automatically, so the install needs network access to GitHub release assets. Do not copy generated contracts into this repository. The PyPI trusted-publishing setup ([#15](https://github.com/Cratis/Chronicle.Python/issues/15)) and first publication ([#16](https://github.com/Cratis/Chronicle.Python/issues/16)) were closed as not planned, so there is -no scheduled move to PyPI. The contracts wheel is generated from the v16.38.2 kernel; see -[Kernel version](#kernel-version) before testing against a newer kernel. +no scheduled move to PyPI. The contracts wheel is generated from the v19.31.3 kernel; see +[Kernel version](#kernel-version) for the kernels it works with. ## Local kernel @@ -41,7 +41,7 @@ built-in development client credentials. Start the development image that matches the contracts version, bound to the loopback interface only: ```shell -docker run --rm -p 127.0.0.1:35000:35000 cratis/chronicle:16.38.2-development +docker run --rm -p 127.0.0.1:35000:35000 cratis/chronicle:19.31.3-development ``` The development image embeds MongoDB inside the container, so every event disappears when the container stops. @@ -57,10 +57,12 @@ configuration contract. ### Kernel version -The contracts and the probe below were exercised against `cratis/chronicle:16.38.2-development`. Newer kernels, -including `latest-development`, may add or change contracts; compatibility between the 16.38.2 contracts and a -later kernel has not been verified. Name the exact kernel image in any issue, test, or pull request that exercises -network behavior. +The contracts, the probe below and the integration tests were exercised against +`cratis/chronicle:19.31.3-development`. The client calls the 19.x contract names (`EventTypes.RegisterEventTypes`, +`EventSequences.Append` in the `Sequences` package), so kernels from before that rename, such as 16.38.2, are not +supported: registering an event type against them fails with `UNIMPLEMENTED`. Kernels newer than 19.31.3, including +`latest-development`, may add or change contracts and have not been verified. Name the exact kernel image in any +issue, test, or pull request that exercises network behavior. ## Authentication contract @@ -85,7 +87,8 @@ call as metadata: authorization: Bearer ``` -Token acquisition, caching, expiry, refresh, and call interception should remain separate from the channel. Do +Token acquisition, caching, expiry, refresh, and call interception are separate from the channel +(`token_provider.py`, `channel.py`). Do not permanently bake one expiring token into channel headers. Chronicle's [authentication and bearer tokens](https://www.cratis.io/chronicle/building-a-client/authentication-and-bearer-tokens/) page describes the behavior the other clients implement: the three authentication modes selected by the connection @@ -100,10 +103,9 @@ validation. Chronicle's .NET client differs: it accepts any server certificate unless validation is turned on; see [TLS configuration](https://www.cratis.io/chronicle/configuration/tls/). This guide does not adopt that default. -How the Python client exposes local-development relaxation (an explicit option, a `skipTlsValidation` -connection-string parameter, or both) is a public API decision for -[connection-string parsing](https://github.com/Cratis/Chronicle.Python/issues/2). Do not decide it implicitly in -an implementation, and do not make relaxed validation the behavior for non-local connections. The connection-string grammar, including `skipTlsValidation`, `apiKey`, and `auth=none`, is in +The Python client reads that decision from the connection string: `skipTlsValidation` is honored only for +loopback hosts, where it trusts the certificate the kernel presents, and is ignored for every other host; see +[Authentication and TLS](authentication-and-tls.md). The connection-string grammar, including `skipTlsValidation`, `apiKey`, and `auth=none`, is in [connection string elements](https://www.cratis.io/chronicle/building-a-client/connection-string-elements/). ### Check the kernel before writing client code @@ -170,7 +172,7 @@ async def main() -> None: asyncio.run(main()) ``` -Run it from the activated development environment. Against `cratis/chronicle:16.38.2-development` it prints: +Run it from the activated development environment. Against `cratis/chronicle:19.31.3-development` it prints: ```text without token: UNAUTHENTICATED @@ -179,7 +181,28 @@ with token: CommandResult The probe creates an event store named `python-probe` in the development kernel. Stopping the container removes it. -## First executable milestone +## Verify against a real kernel + +The unit tests run an in-process fake kernel. The opt-in integration tests need a development kernel you start +yourself. Use a private port so they cannot collide with other kernels: + +```shell +docker run -d --name chronicle-python-it -p 127.0.0.1:19300:35000 cratis/chronicle:19.31.3-development +# wait until the log says "ready and listening on port 35000" +CHRONICLE_INTEGRATION_URL=chronicle://localhost:19300 pytest tests/test_integration.py --no-cov +docker rm -f chronicle-python-it +``` + +The tests authenticate over TLS, ensure an event store and the `Default` namespace, register an event type, append +two events and assert consecutive sequence numbers. They also assert that wrong credentials fail with +`TokenAuthorizationError` without leaking the secret, and that `skipTlsValidation=false` rejects the self-signed +certificate. Without `CHRONICLE_INTEGRATION_URL` they are skipped. The kernel also keeps its data in the container, +so removing it cleans up. + +The client targets the 19.x contract names. Against a 16.x kernel, ensuring an event store and a namespace works, +but event type registration returns `UNIMPLEMENTED`. + +## First executable milestone (implemented) Implement and verify this order before expanding the API: @@ -238,7 +261,7 @@ uncertainty against the core contracts and kernel behavior. | Symptom | Likely cause and fix | | --- | --- | -| `pip install -e ".[dev]"` fails while downloading `cratis_chronicle_contracts-16.38.2-py3-none-any.whl` | The install cannot reach GitHub release assets. Allow `github.com` and `release-assets.githubusercontent.com`, where the download redirects, through your proxy or firewall | +| `pip install -e ".[dev]"` fails while downloading `cratis_chronicle_contracts-19.31.3-py3-none-any.whl` | The install cannot reach GitHub release assets. Allow `github.com` and `release-assets.githubusercontent.com`, where the download redirects, through your proxy or firewall | | `pip` reports that hashes do not match | The downloaded wheel differs from the pinned SHA-256. Do not remove the hash; report it in an issue | | `docker run` fails with `port is already allocated` | Another Chronicle kernel or process uses port 35000. Stop it, or publish a different host port (`-p 127.0.0.1:35100:35000`) and use that port in the connection string, the `curl` URL, and the probe's `PORT` | | `ssl.SSLEOFError` or a refused connection right after `docker run` | The kernel is still starting. Wait until the token check succeeds, then retry | @@ -247,8 +270,8 @@ uncertainty against the core contracts and kernel behavior. ## Next steps -- Pick up [async OAuth token handling](https://github.com/Cratis/Chronicle.Python/issues/3); connection-string - parsing is described in [Connection strings](connection-strings.md). +- Read [Getting started](getting-started.md), [Authentication and TLS](authentication-and-tls.md) and + [Connection strings](connection-strings.md). - Read Chronicle's [Building a Chronicle client](https://www.cratis.io/chronicle/building-a-client/) guide for the cross-client contract. - Follow [CONTRIBUTING.md](../CONTRIBUTING.md) for the required checks before opening a pull request. diff --git a/Documentation/connection-strings.md b/Documentation/connection-strings.md index 68a8ff9..25c59f6 100644 --- a/Documentation/connection-strings.md +++ b/Documentation/connection-strings.md @@ -44,10 +44,13 @@ Percent-encode a `:` in the client secret as `%3A`; an unencoded one raises `Inc ## TLS certificate validation -`skipTlsValidation` is a boolean that defaults to `true`, as in the .NET client: the connection always uses TLS, but the -kernel certificate is accepted without validation so a development kernel with a self-signed certificate works. -Pass `skipTlsValidation=false` to require a verifiable certificate. `true` and `false` are accepted; any other value -raises `UnsupportedOptionError`. The result is `options.skip_tls_validation`; `options.tls` stays `True` either way. +`skipTlsValidation` is a boolean that defaults to `true`, as in the .NET client. The connection always uses TLS. The +client honors the option only for `localhost` and loopback addresses, where it trusts the certificate the local +kernel presents so a development kernel with a self-signed certificate works. For any other host the certificate must +chain to a trusted root whatever the option says. Pass `skipTlsValidation=false` to require a verifiable certificate +everywhere. `true` and `false` are accepted; any other value raises `UnsupportedOptionError`. The result is +`options.skip_tls_validation`; `options.tls` stays `True` either way. See +[Authentication and TLS](authentication-and-tls.md). ```python parse_connection_string("chronicle://localhost:35000/?skipTlsValidation=false").skip_tls_validation # False diff --git a/Documentation/getting-started.md b/Documentation/getting-started.md index a212412..148384e 100644 --- a/Documentation/getting-started.md +++ b/Documentation/getting-started.md @@ -1,21 +1,103 @@ -# Getting started +--- +title: Getting started +description: Install the experimental Chronicle Python client from a source checkout, authenticate to a local development kernel and append your first event. +--- -Chronicle.Python does not yet expose a usable client API or published package. This page will become the -installation and first-append guide when the -[initial authenticated append milestone](https://github.com/Cratis/Chronicle.Python/issues/4) passes its tests. +This client is experimental. Its API changes without notice, nothing is published to PyPI, and it supports one +workflow: connect, authenticate, ensure an event store and namespace, register an event type and append an event. +Projections, reducers, reactors, subscriptions, reconnect handling and automatic kernel discovery are not +implemented. + +## Install from a source checkout + +Python 3.10 or newer is required. The install downloads the generated contracts wheel from GitHub release assets. + +```shell +git clone https://github.com/Cratis/Chronicle.Python.git +cd Chronicle.Python +python -m venv .venv +source .venv/bin/activate +python -m pip install -e ".[dev]" +``` + +## Start a development kernel + +The contracts in this client are generated from Chronicle 19.31.3, so use the matching kernel image. The development +image generates a self-signed certificate, accepts the built-in development credentials and keeps events inside the +container. + +```shell +docker run --rm -p 127.0.0.1:35000:35000 cratis/chronicle:19.31.3-development +``` + +:::caution[Older kernels are not supported] +The client calls the 19.x gRPC contracts. Against a 16.x kernel, such as `cratis/chronicle:16.38.2-development`, +ensuring an event store and a namespace works but registering an event type fails with `UNIMPLEMENTED`, because the +kernel renamed that contract. Use a 19.31.3 or compatible kernel. +::: + +## Append an event + +```python +import asyncio +import uuid + +from cratis_chronicle import ChronicleClient, EventTypeDefinition + +BOOK_ADDED = EventTypeDefinition( + id="book-added", + schema={ + "type": "object", + "properties": {"title": {"type": "string"}, "isbn": {"type": "string"}}, + "required": ["title", "isbn"], + }, +) + + +async def main() -> None: + async with await ChronicleClient.connect("chronicle://localhost:35000") as client: + event_store = await client.ensure_event_store("library") + namespace = await event_store.ensure_namespace("Default") + await event_store.register_event_type(BOOK_ADDED) + + result = await namespace.event_log.append( + event_source_id=str(uuid.uuid4()), + event_type=BOOK_ADDED, + content={"title": "Event Sourcing in Python", "isbn": "978-0-00-000000-0"}, + ) + print(result.sequence_number) + + +asyncio.run(main()) +``` + +The same program ships as [`Samples/append_event/main.py`](../Samples/README.md). Leaving the `async with` block +closes the gRPC channel and the token provider. + +## Authentication and TLS + +The client requests an OAuth token with the credentials from the connection string and attaches it to every new gRPC +call. See [Connection strings](connection-strings.md) for the credentials, and +[Authentication and TLS](authentication-and-tls.md) for token refresh, certificate trust and error handling. + +## Errors + +| Error | Meaning | +| --- | --- | +| `TokenAuthorizationError` | The token endpoint rejected the client credentials | +| `TokenRequestError` | The token request failed: network, TLS, timeout or an unexpected status | +| `TokenResponseError` | The token endpoint returned something that is not a token response | +| `CommandFailedError` | The kernel reported a failure while ensuring an event store or namespace | +| `AppendFailedError` | The kernel did not append the event | + +All of them derive from `ChronicleError`. No message contains the client secret or an access token. ## What works today | You want to… | Status | | --- | --- | | `pip install cratis-chronicle` from PyPI | Not possible. No package is published | -| Connect to Chronicle and append events from Python | Not possible through this package yet. It exposes `__version__` and a connection-string parser that opens no connection | -| Parse a `chronicle://` connection string | Supported. See [Connection strings](connection-strings.md) | -| Build the client from source and run its checks | Supported. See [Development setup](../README.md#development-setup) | +| Authenticate, ensure state, register an event type and append | Supported against a 19.31.3 development kernel | +| Read events, observe, project or react | Not implemented | +| Reconnect or discover a kernel automatically | Not implemented | | Use Chronicle from another language now | Use the [.NET](https://github.com/Cratis/Chronicle), [TypeScript](https://github.com/Cratis/Chronicle.TypeScript), [Kotlin/Java](https://github.com/Cratis/Chronicle.Kotlin), or [Elixir](https://github.com/Cratis/Chronicle.Elixir) client | - -## Contribute - -To contribute now, follow [Building the Chronicle Python client](client-development-guide.md) and the repository -[contribution guide](../CONTRIBUTING.md). The development guide shows how to run a local kernel and check the -token and gRPC path before you write client code. diff --git a/README.md b/README.md index 25939d5..b277470 100644 --- a/README.md +++ b/README.md @@ -11,17 +11,18 @@ contracts. Python joins the existing [.NET](https://github.com/Cratis/Chronicle) and [Elixir](https://github.com/Cratis/Chronicle.Elixir) Chronicle clients. > [!IMPORTANT] -> The client is in its initial implementation stage. It has no usable client API yet: the `cratis_chronicle` -> package exposes `__version__` and a connection-string parser that opens no connection. Nothing is published to -> PyPI, and no compatibility, feature-parity, or support commitment is implied. +> The client is in its initial implementation stage. The `cratis_chronicle` package can authenticate to a local +> development kernel, ensure an event store and namespace, register an event type and append an event; nothing else +> is implemented. Nothing is published to PyPI, and no compatibility, feature-parity, or support commitment is +> implied. ## Current status | Area | Status | | --- | --- | -| Client API (connect, authenticate, append) | Not implemented. Tracked by [the first authenticated append milestone](https://github.com/Cratis/Chronicle.Python/issues/4) | +| Client API (connect, authenticate, append) | Experimental and minimal: OAuth, event store and namespace, event type registration, append to `event-log`. Verified against the 19.31.3 development kernel. See [Getting started](Documentation/getting-started.md) | | `cratis-chronicle` on PyPI | Not published. Install from a source checkout | -| Generated contracts (`cratis-chronicle-contracts`) | Not on PyPI. Installed automatically from a SHA-256-pinned wheel attached to the [Chronicle v16.38.2 release](https://github.com/Cratis/Chronicle/releases/tag/v16.38.2) | +| Generated contracts (`cratis-chronicle-contracts`) | Not on PyPI. Installed automatically from a SHA-256-pinned wheel attached to the [Chronicle v19.31.3 release](https://github.com/Cratis/Chronicle/releases/tag/v19.31.3) | | Python versions | 3.10 or newer; CI runs 3.10, 3.11, 3.12, 3.13, and 3.14 | | Shared Chronicle documentation (language tabs) | Not integrated. Tracked by [Python examples in shared Chronicle documentation](https://github.com/Cratis/Chronicle.Python/issues/5) | @@ -33,7 +34,7 @@ Chronicle kernel. 1. Read [CONTRIBUTING.md](CONTRIBUTING.md). 2. Follow the [client development guide](Documentation/client-development-guide.md). -3. Start with [async OAuth token handling](https://github.com/Cratis/Chronicle.Python/issues/3), the next step after connection-string parsing. +3. Pick a [parity issue](https://github.com/Cratis/Chronicle.Python/issues), such as reading events or observers. 4. Submit a pull request from a fork or branch. ## Development setup diff --git a/Samples/README.md b/Samples/README.md index 4e2f057..831afae 100644 --- a/Samples/README.md +++ b/Samples/README.md @@ -1,7 +1,18 @@ # Samples -Runnable samples will be added with the client milestones they exercise. A sample must use the released or local -client package, contain no credentials beyond documented local-development defaults, and run in CI before it is -presented as working. +A sample uses the local client package, contains no credentials beyond the documented local-development defaults, and +is run against a development kernel before it is presented as working. -The first planned sample will authenticate to a local Chronicle development kernel and append one event. +## append_event + +[`append_event/main.py`](append_event/main.py) authenticates to a local development kernel, ensures an event store and +the `Default` namespace, registers an event type with a JSON schema and appends one event. + +```shell +docker run -d --name chronicle-python-sample -p 127.0.0.1:35000:35000 cratis/chronicle:19.31.3-development +python Samples/append_event/main.py # chronicle://localhost:35000 +python Samples/append_event/main.py chronicle://localhost:19300 # another port +docker rm -f chronicle-python-sample +``` + +It prints `Appended event with sequence number `. diff --git a/Samples/append_event/main.py b/Samples/append_event/main.py new file mode 100644 index 0000000..7f1755a --- /dev/null +++ b/Samples/append_event/main.py @@ -0,0 +1,43 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Authenticate to a local Chronicle development kernel, register an event type and append one event. + +Run ``python Samples/append_event/main.py [connection-string]``. The default connection string targets a development +kernel on localhost:35000 with the built-in development credentials. +""" + +import asyncio +import sys +import uuid + +from cratis_chronicle import ChronicleClient, EventTypeDefinition + +DEFAULT_CONNECTION_STRING = "chronicle://localhost:35000" + +BOOK_ADDED = EventTypeDefinition( + id="python-sample.book-added", + schema={ + "type": "object", + "properties": {"title": {"type": "string"}, "isbn": {"type": "string"}}, + "required": ["title", "isbn"], + }, +) + + +async def main(connection_string: str) -> None: + async with await ChronicleClient.connect(connection_string) as client: + event_store = await client.ensure_event_store("python-sample") + namespace = await event_store.ensure_namespace("Default") + await event_store.register_event_type(BOOK_ADDED) + + result = await namespace.event_log.append( + event_source_id=str(uuid.uuid4()), + event_type=BOOK_ADDED, + content={"title": "Event Sourcing in Python", "isbn": "978-0-00-000000-0"}, + ) + print(f"Appended event with sequence number {result.sequence_number}") + + +if __name__ == "__main__": + asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONNECTION_STRING)) diff --git a/pyproject.toml b/pyproject.toml index dcc2b2b..e933b43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,8 +36,9 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ + "grpcio>=1.83.0,<2", # Temporary release asset until PyPI trusted publishing is configured. - "cratis-chronicle-contracts @ https://github.com/Cratis/Chronicle/releases/download/v16.38.2/cratis_chronicle_contracts-16.38.2-py3-none-any.whl#sha256=357052da8a62653dcfa40c8051c4af2c6a6d29295cb6783f59dd5efdf4bdc3f9", + "cratis-chronicle-contracts @ https://github.com/Cratis/Chronicle/releases/download/v19.31.3/cratis_chronicle_contracts-19.31.3-py3-none-any.whl#sha256=fcdc4571947f218d59af0ec56ae532032cf3aad2ee504292730735f6cef56fca", ] [project.optional-dependencies] @@ -71,6 +72,7 @@ packages = ["src/cratis_chronicle"] [tool.pytest.ini_options] addopts = "--strict-config --strict-markers --cov=cratis_chronicle --cov-report=term-missing" testpaths = ["tests"] +markers = ["integration: needs a real Chronicle development kernel; set CHRONICLE_INTEGRATION_URL"] [tool.ruff] line-length = 120 @@ -83,3 +85,9 @@ select = ["B", "E", "F", "I", "UP"] python_version = "3.10" strict = true packages = ["cratis_chronicle"] + +[[tool.mypy.overrides]] +# The generated contracts ship stubs that mypy cannot parse, and grpcio ships none. Both are used through narrow wrappers. +module = ["cratis_chronicle_contracts.*", "grpc", "grpc.*"] +follow_imports = "skip" +ignore_missing_imports = true diff --git a/src/cratis_chronicle/__init__.py b/src/cratis_chronicle/__init__.py index f8e9493..d148c80 100644 --- a/src/cratis_chronicle/__init__.py +++ b/src/cratis_chronicle/__init__.py @@ -5,6 +5,8 @@ from importlib.metadata import PackageNotFoundError, version +from .channel import BearerTokenInterceptor, ChronicleChannel +from .client import DEFAULT_NAMESPACE, EVENT_LOG, ChronicleClient, EventSequence, EventStore, Namespace from .connection_string import ( DEFAULT_PORT, DEVELOPMENT_CLIENT_ID, @@ -22,6 +24,21 @@ UnsupportedSchemeError, parse_connection_string, ) +from .errors import ( + AppendFailedError, + ChronicleError, + CommandFailedError, + InsecureTransportError, + TokenAuthorizationError, + TokenError, + TokenRequestError, + TokenResponseError, +) +from .events import AppendResult, EventTypeDefinition +from .http_transport import HttpResponse, StreamFormTransport, TokenTransport +from .tls import TlsTrust, is_loopback_host, resolve_tls_trust +from .token_provider import DEFAULT_TOKEN_LIFETIME_SECONDS, OAuthTokenProvider +from .wire import ConceptAs try: __version__ = version("cratis-chronicle") @@ -29,6 +46,33 @@ __version__ = "0.0.0" __all__ = [ + "DEFAULT_NAMESPACE", + "EVENT_LOG", + "DEFAULT_TOKEN_LIFETIME_SECONDS", + "AppendFailedError", + "AppendResult", + "BearerTokenInterceptor", + "ChronicleChannel", + "ChronicleClient", + "ChronicleError", + "CommandFailedError", + "ConceptAs", + "EventSequence", + "EventStore", + "EventTypeDefinition", + "HttpResponse", + "Namespace", + "OAuthTokenProvider", + "StreamFormTransport", + "TlsTrust", + "TokenAuthorizationError", + "InsecureTransportError", + "TokenError", + "TokenRequestError", + "TokenResponseError", + "TokenTransport", + "is_loopback_host", + "resolve_tls_trust", "DEFAULT_PORT", "DEVELOPMENT_CLIENT_ID", "DEVELOPMENT_CLIENT_SECRET", diff --git a/src/cratis_chronicle/_event_type_contracts.py b/src/cratis_chronicle/_event_type_contracts.py new file mode 100644 index 0000000..6ba5058 --- /dev/null +++ b/src/cratis_chronicle/_event_type_contracts.py @@ -0,0 +1,16 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Dynamic access to the generated event type contracts. + +The ``eventtypes_pb2.pyi`` stub in the contracts package declares an enum member named ``None``, which is not valid +Python syntax, so mypy aborts as soon as it parses the stub. Importing the modules dynamically keeps the type check +running; the values are used through the narrow wrappers in ``client.py``. Remove this module when the contracts stub +is valid. +""" + +from importlib import import_module +from typing import Any + +messages: Any = import_module("cratis_chronicle_contracts.eventtypes_pb2") +services: Any = import_module("cratis_chronicle_contracts.eventtypes_pb2_grpc") diff --git a/src/cratis_chronicle/channel.py b/src/cratis_chronicle/channel.py new file mode 100644 index 0000000..413e787 --- /dev/null +++ b/src/cratis_chronicle/channel.py @@ -0,0 +1,95 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""The authenticated gRPC channel. Token acquisition stays in the token provider; the channel only asks it per call.""" + +from __future__ import annotations + +import grpc +from grpc import aio + +from .connection_string import ChronicleConnectionOptions +from .errors import InsecureTransportError +from .http_transport import StreamFormTransport, TokenTransport +from .tls import is_loopback_host, resolve_tls_trust +from .token_provider import OAuthTokenProvider + +__all__ = ["BearerTokenInterceptor", "ChronicleChannel"] + + +def _with_bearer(details: aio.ClientCallDetails, token: str) -> aio.ClientCallDetails: + metadata = aio.Metadata(*(details.metadata or ())) + metadata.add("authorization", f"Bearer {token}") + return aio.ClientCallDetails(details.method, details.timeout, metadata, details.credentials, details.wait_for_ready) + + +class BearerTokenInterceptor(aio.UnaryUnaryClientInterceptor, aio.UnaryStreamClientInterceptor): # type: ignore[misc] + """Adds ``authorization: Bearer `` to every new call with the provider's current token. + + The token is looked up when each call starts, never baked into the channel, so refreshes take effect on the next + call. A token failure surfaces as the provider's ``TokenError``. + """ + + def __init__(self, provider: OAuthTokenProvider) -> None: + self._provider = provider + + async def intercept_unary_unary(self, continuation, client_call_details, request): # type: ignore[no-untyped-def] + return await continuation(_with_bearer(client_call_details, await self._provider.get_token()), request) + + async def intercept_unary_stream(self, continuation, client_call_details, request): # type: ignore[no-untyped-def] + return await continuation(_with_bearer(client_call_details, await self._provider.get_token()), request) + + +class ChronicleChannel: + """Owns the gRPC channel and the token provider and closes both deterministically.""" + + def __init__(self, channel: aio.Channel, provider: OAuthTokenProvider) -> None: + self._channel = channel + self._provider = provider + self._closed = False + + @property + def channel(self) -> aio.Channel: + """The underlying authenticated ``grpc.aio`` channel, for generated stubs.""" + return self._channel + + @classmethod + async def open( + cls, + options: ChronicleConnectionOptions, + *, + ca_certificates: bytes | None = None, + token_transport: TokenTransport | None = None, + ) -> ChronicleChannel: + """Create the channel for ``options``. Nothing is sent until the first call; the first call fetches a token.""" + if not options.tls and not is_loopback_host(options.host): + raise InsecureTransportError( + "Client credentials and access tokens are not sent without TLS to a host other than localhost" + ) + trust = await resolve_tls_trust(options, ca_certificates=ca_certificates) + transport = token_transport or StreamFormTransport(trust.ssl_context() if trust else None) + provider = OAuthTokenProvider(options, transport) + target = f"{options.host}:{options.port}" if ":" not in options.host else f"[{options.host}]:{options.port}" + interceptors = [BearerTokenInterceptor(provider)] + if trust is None: + channel = aio.insecure_channel(target, interceptors=interceptors) + else: + credentials = grpc.ssl_channel_credentials(root_certificates=trust.root_certificates) + channel = aio.secure_channel(target, credentials, interceptors=interceptors) + return cls(channel, provider) + + async def aclose(self) -> None: + """Close the channel and the token provider. Safe to call more than once.""" + if self._closed: + return + self._closed = True + try: + await self._channel.close() + finally: + await self._provider.aclose() + + async def __aenter__(self) -> ChronicleChannel: + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.aclose() diff --git a/src/cratis_chronicle/client.py b/src/cratis_chronicle/client.py new file mode 100644 index 0000000..fe632bd --- /dev/null +++ b/src/cratis_chronicle/client.py @@ -0,0 +1,188 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""The Chronicle client: connect, ensure an event store and namespace, register an event type and append.""" + +from __future__ import annotations + +import datetime +import uuid +from collections.abc import Iterable +from typing import Any + +from cratis_chronicle_contracts import sequences_pb2 as sequence_messages +from cratis_chronicle_contracts.eventstores_pb2 import EnsureEventStoreRequest +from cratis_chronicle_contracts.eventstores_pb2_grpc import EventStoresStub +from cratis_chronicle_contracts.namespaces_pb2 import EnsureNamespaceRequest +from cratis_chronicle_contracts.namespaces_pb2_grpc import NamespacesStub +from cratis_chronicle_contracts.sequences_pb2_grpc import EventSequencesStub + +from . import _event_type_contracts as event_type_contracts +from .channel import ChronicleChannel +from .connection_string import ChronicleConnectionOptions, parse_connection_string +from .errors import AppendFailedError, CommandFailedError +from .events import AppendResult, EventTypeDefinition +from .wire import from_wire_guid, serialize_content, to_wire_guid, to_wire_timestamp + +__all__ = ["DEFAULT_NAMESPACE", "EVENT_LOG", "ChronicleClient", "EventSequence", "EventStore", "Namespace"] + +DEFAULT_NAMESPACE = "Default" +EVENT_LOG = "event-log" +_ERROR_SEVERITY = 3 +UNAVAILABLE_SEQUENCE_NUMBER = 2**64 - 1 +CAUSATION_TYPE = "Python Client" +IDENTITY = "cratis-chronicle-python" + + +def _ensure_command_succeeded(operation: str, result: Any) -> None: + reasons = [message for message in result.ExceptionMessages] + reasons += [validation.Message for validation in result.ValidationResults if validation.Severity == _ERROR_SEVERITY] + # IsAuthorized is not consulted: the contract defaults it to true, so the kernel omits it for success and proto3 + # reads that as false. An authorization failure carries its reason. + if result.AuthorizationFailureReason: + reasons.append(result.AuthorizationFailureReason) + if reasons: + raise CommandFailedError(operation, reasons) + + +class EventSequence: + """An event sequence of one namespace, such as the event log.""" + + def __init__(self, stub: Any, event_store: str, namespace: str, sequence_id: str) -> None: + self._stub = stub + self._event_store = event_store + self._namespace = namespace + self._sequence_id = sequence_id + + async def append( + self, + event_source_id: str, + event_type: EventTypeDefinition, + content: Any, + *, + occurred: datetime.datetime | None = None, + ) -> AppendResult: + """Append one event and return its sequence number. + + ``content`` is a mapping or dataclass whose property names must match the registered schema. The event type + must have been registered with :meth:`EventStore.register_event_type`. + + Raises: + AppendFailedError: The kernel reported errors, a constraint violation or a concurrency violation. + """ + if not event_source_id.strip(): + raise ValueError("An event source id must not be empty") + correlation_id = uuid.uuid4() + timestamp = sequence_messages.SerializableDateTimeOffset( + Value=to_wire_timestamp(occurred or datetime.datetime.now(datetime.timezone.utc)) + ) + request = sequence_messages.AppendRequest( + EventStore=self._event_store, + Namespace=self._namespace, + EventSequenceId=self._sequence_id, + CorrelationId=to_wire_guid(correlation_id), + EventSourceId=event_source_id, + EventType=sequence_messages.EventType(Id=event_type.id, Generation=event_type.generation), + Content=serialize_content(content), + # The kernel dereferences the causation chain, so an empty chain fails the call; record who appended. + Causation=[sequence_messages.Causation(Occurred=timestamp, Type=CAUSATION_TYPE)], + CausedBy=sequence_messages.Identity(Subject=IDENTITY, Name=IDENTITY, UserName=IDENTITY), + Occurred=timestamp, + ) + # An unset scope makes the kernel dereference null, and sequence number 0 would demand an empty sequence. + # The unavailable sequence number is the value the kernel does not validate. + request.ConcurrencyScope.SequenceNumber = UNAVAILABLE_SEQUENCE_NUMBER + result = await self._stub.Append(request) + _ensure_command_succeeded("Appending an event", result) + response = result.Response + reasons = list(response.Errors) + reasons += [f"constraint violation: {violation.Message}" for violation in response.ConstraintViolations] + if response.HasField("ConcurrencyViolation"): + reasons.append("concurrency violation") + if reasons: + raise AppendFailedError(reasons) + return AppendResult( + sequence_number=response.SequenceNumber, correlation_id=from_wire_guid(response.CorrelationId) + ) + + +class Namespace: + """A namespace of an event store.""" + + def __init__(self, channel: ChronicleChannel, event_store: str, name: str) -> None: + self.name = name + self._event_log = EventSequence(EventSequencesStub(channel.channel), event_store, name, EVENT_LOG) + + @property + def event_log(self) -> EventSequence: + """The default event sequence.""" + return self._event_log + + +class EventStore: + """An event store that exists in the kernel.""" + + def __init__(self, channel: ChronicleChannel, name: str) -> None: + self.name = name + self._channel = channel + + async def ensure_namespace(self, name: str = DEFAULT_NAMESPACE) -> Namespace: + """Create the namespace when it does not exist yet and return it.""" + result = await NamespacesStub(self._channel.channel).EnsureNamespace( + EnsureNamespaceRequest(EventStore=self.name, Namespace=name) + ) + _ensure_command_succeeded(f"Ensuring namespace '{name}'", result) + return Namespace(self._channel, self.name, name) + + async def register_event_types(self, definitions: Iterable[EventTypeDefinition]) -> None: + """Register event types with their schemas in this event store.""" + messages = event_type_contracts.messages + registrations = [ + messages.EventTypeRegistration( + Type=messages.EventType(Id=definition.id, Generation=definition.generation), + Schema=definition.schema_json(), + ) + for definition in definitions + ] + result = await event_type_contracts.services.EventTypesStub(self._channel.channel).RegisterEventTypes( + messages.RegisterEventTypesRequest(EventStore=self.name, Types=registrations) + ) + _ensure_command_succeeded("Registering event types", result) + + async def register_event_type(self, definition: EventTypeDefinition) -> None: + """Register one event type with its schema in this event store.""" + await self.register_event_types([definition]) + + +class ChronicleClient: + """An experimental async client for a Chronicle kernel. Use it as an async context manager.""" + + def __init__(self, channel: ChronicleChannel) -> None: + self._channel = channel + + @classmethod + async def connect( + cls, + connection: str | ChronicleConnectionOptions, + *, + ca_certificates: bytes | None = None, + ) -> ChronicleClient: + """Create a client from a connection string or options. See :class:`ChronicleChannel` for the TLS rules.""" + options = parse_connection_string(connection) if isinstance(connection, str) else connection + return cls(await ChronicleChannel.open(options, ca_certificates=ca_certificates)) + + async def ensure_event_store(self, name: str) -> EventStore: + """Create the event store when it does not exist yet and return it.""" + result = await EventStoresStub(self._channel.channel).EnsureEventStore(EnsureEventStoreRequest(Name=name)) + _ensure_command_succeeded(f"Ensuring event store '{name}'", result) + return EventStore(self._channel, name) + + async def aclose(self) -> None: + """Close the channel and the token provider. Safe to call more than once.""" + await self._channel.aclose() + + async def __aenter__(self) -> ChronicleClient: + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.aclose() diff --git a/src/cratis_chronicle/errors.py b/src/cratis_chronicle/errors.py new file mode 100644 index 0000000..725dd92 --- /dev/null +++ b/src/cratis_chronicle/errors.py @@ -0,0 +1,74 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Exceptions raised by the Chronicle client.""" + +from __future__ import annotations + +__all__ = [ + "AppendFailedError", + "ChronicleError", + "CommandFailedError", + "InsecureTransportError", + "TokenAuthorizationError", + "TokenError", + "TokenRequestError", + "TokenResponseError", +] + + +class ChronicleError(Exception): + """Base class for every error the client raises on its own behalf. + + Messages never contain the client secret or an access token. + """ + + +class CommandFailedError(ChronicleError): + """The kernel answered a command with a failure result.""" + + def __init__(self, operation: str, reasons: list[str]) -> None: + self.operation = operation + self.reasons = reasons + super().__init__(f"{operation} failed: {'; '.join(reasons) if reasons else 'no reason given'}") + + +class AppendFailedError(ChronicleError): + """The kernel did not append the event: it reported errors, a constraint violation or a concurrency violation.""" + + def __init__(self, reasons: list[str]) -> None: + self.reasons = reasons + super().__init__(f"The event was not appended: {'; '.join(reasons) if reasons else 'no reason given'}") + + +class InsecureTransportError(ChronicleError): + """The options would send the client secret or an access token over an unencrypted connection to a remote host.""" + + +class TokenError(ChronicleError): + """Base class for failures to obtain an access token.""" + + +class TokenRequestError(TokenError): + """The token request could not be completed: a network, TLS or timeout failure, or an unexpected HTTP status.""" + + def __init__(self, message: str, *, status: int | None = None) -> None: + self.status = status + super().__init__(message) + + +class TokenAuthorizationError(TokenError): + """The token endpoint rejected the client credentials, for example with an OAuth ``invalid_client`` error.""" + + def __init__(self, status: int, error: str | None, description: str | None) -> None: + self.status = status + self.error = error + self.description = description + detail = ": ".join(part for part in (error, description) if part) + super().__init__( + f"The token endpoint rejected the client credentials (HTTP {status}){': ' + detail if detail else ''}" + ) + + +class TokenResponseError(TokenError): + """The token endpoint answered successfully but the response is not a usable token response.""" diff --git a/src/cratis_chronicle/events.py b/src/cratis_chronicle/events.py new file mode 100644 index 0000000..f5bfd13 --- /dev/null +++ b/src/cratis_chronicle/events.py @@ -0,0 +1,46 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Event type definitions and append results.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any +from uuid import UUID + +__all__ = ["AppendResult", "EventTypeDefinition"] + + +@dataclass(frozen=True, slots=True) +class EventTypeDefinition: + """An event type to register: its identifier, generation and JSON schema. + + The schema must be a non-empty JSON schema object. The kernel validates appended content against it. + """ + + id: str + schema: Mapping[str, Any] + generation: int = 1 + + def __post_init__(self) -> None: + if not self.id.strip(): + raise ValueError("An event type id must not be empty") + if self.generation < 1: + raise ValueError("An event type generation must be 1 or greater") + if not self.schema: + raise ValueError("An event type needs a non-empty JSON schema") + + def schema_json(self) -> str: + """The schema serialized for the wire.""" + return json.dumps(self.schema, separators=(",", ":"), allow_nan=False) + + +@dataclass(frozen=True, slots=True) +class AppendResult: + """The outcome of a successful append.""" + + sequence_number: int + correlation_id: UUID diff --git a/src/cratis_chronicle/http_transport.py b/src/cratis_chronicle/http_transport.py new file mode 100644 index 0000000..66c3833 --- /dev/null +++ b/src/cratis_chronicle/http_transport.py @@ -0,0 +1,137 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""A minimal asyncio HTTP/1.1 form-POST transport for the OAuth token endpoint. + +It exists so a request can be cancelled for real and so the client needs no HTTP dependency. It follows no redirects, +which keeps the client secret from ever being sent to a host the caller did not configure. +""" + +from __future__ import annotations + +import asyncio +import ssl +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Protocol +from urllib.parse import urlencode, urlsplit + +from .errors import TokenRequestError + +__all__ = ["HttpResponse", "StreamFormTransport", "TokenTransport"] + +_MAX_RESPONSE_BYTES = 1024 * 1024 + + +@dataclass(frozen=True, slots=True) +class HttpResponse: + """The status and body of an HTTP response.""" + + status: int + body: bytes + + +class TokenTransport(Protocol): + """Sends the token request. Replaceable so the provider can be tested without a network.""" + + async def post_form(self, url: str, fields: Mapping[str, str]) -> HttpResponse: + """POST ``fields`` as ``application/x-www-form-urlencoded`` to ``url`` and return the response.""" + ... + + +class StreamFormTransport: + """Sends the request over a fresh asyncio connection that is closed after the response.""" + + def __init__(self, ssl_context: ssl.SSLContext | None, *, timeout: float = 30.0) -> None: + self._ssl_context = ssl_context + self._timeout = timeout + + async def post_form(self, url: str, fields: Mapping[str, str]) -> HttpResponse: + parts = urlsplit(url) + secure = parts.scheme == "https" + host = parts.hostname or "" + port = parts.port or (443 if secure else 80) + body = urlencode(fields).encode("ascii") + host_header = parts.netloc.rpartition("@")[2] + head = ( + f"POST {parts.path or '/'} HTTP/1.1\r\nHost: {host_header}\r\n" + "Content-Type: application/x-www-form-urlencoded\r\nAccept: application/json\r\n" + f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n" + ).encode("ascii") + try: + return await asyncio.wait_for(self._exchange(host, port, secure, head + body), self._timeout) + except (TimeoutError, asyncio.TimeoutError) as error: + raise TokenRequestError("The token request timed out") from error + except ssl.SSLError as error: + raise TokenRequestError( + f"The TLS handshake with the token endpoint failed: {error.reason or 'ssl error'}" + ) from error + except (OSError, EOFError, ValueError) as error: + raise TokenRequestError(f"The token endpoint could not be reached ({type(error).__name__})") from error + + async def _exchange(self, host: str, port: int, secure: bool, payload: bytes) -> HttpResponse: + reader, writer = await asyncio.open_connection( + host, + port, + ssl=(self._ssl_context or ssl.create_default_context()) if secure else None, + server_hostname=host if secure else None, + ) + try: + writer.write(payload) + await writer.drain() + return _parse_response(await _read_to_end(reader)) + finally: + writer.close() + try: + await writer.wait_closed() + except (OSError, ssl.SSLError): + pass + + +async def _read_to_end(reader: asyncio.StreamReader) -> bytes: + """Read until the server closes the connection (the request said ``Connection: close``), within the size bound.""" + raw = bytearray() + while chunk := await reader.read(65536): + raw += chunk + if len(raw) > _MAX_RESPONSE_BYTES: + raise ValueError("response too large") + return bytes(raw) + + +def _parse_response(raw: bytes) -> HttpResponse: + head, separator, body = raw.partition(b"\r\n\r\n") + if not separator: + raise ValueError("incomplete response") + lines = head.decode("latin-1").split("\r\n") + status_parts = lines[0].split(" ", 2) + if len(status_parts) < 2 or not status_parts[1].isdigit(): + raise ValueError("invalid status line") + headers = {} + for line in lines[1:]: + name, _, value = line.partition(":") + headers[name.strip().lower()] = value.strip().lower() + if "chunked" in headers.get("transfer-encoding", ""): + body = _decode_chunked(body) + elif "content-length" in headers: + length = headers["content-length"] + if not length.isascii() or not length.isdigit(): + raise ValueError("invalid content length") + if len(body) < int(length): + raise ValueError("truncated body") + body = body[: int(length)] + return HttpResponse(status=int(status_parts[1]), body=body) + + +def _decode_chunked(data: bytes) -> bytes: + decoded = bytearray() + while True: + size_line, separator, data = data.partition(b"\r\n") + if not separator: + raise ValueError("invalid chunked body") + size = int(size_line.split(b";", 1)[0], 16) + if size == 0: + return bytes(decoded) + if len(data) < size + 2: + raise ValueError("truncated chunked body") + decoded += data[:size] + data = data[size + 2 :] diff --git a/src/cratis_chronicle/tls.py b/src/cratis_chronicle/tls.py new file mode 100644 index 0000000..decc5af --- /dev/null +++ b/src/cratis_chronicle/tls.py @@ -0,0 +1,77 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Decides which certificates the client trusts. Production connections always verify the certificate and host name.""" + +from __future__ import annotations + +import asyncio +import ipaddress +import ssl +from collections.abc import Callable +from dataclasses import dataclass + +from .connection_string import ChronicleConnectionOptions + +__all__ = ["TlsTrust", "is_loopback_host", "resolve_tls_trust"] + +CertificateFetcher = Callable[[str, int], str] + + +@dataclass(frozen=True, slots=True) +class TlsTrust: + """The trusted root certificates, as PEM, or ``None`` to use the platform's default roots. + + Certificate and host name verification stay on in both cases. + """ + + root_certificates: bytes | None + + def ssl_context(self) -> ssl.SSLContext: + """Create a verifying SSL context that trusts exactly these roots (or the platform defaults).""" + if self.root_certificates is None: + return ssl.create_default_context() + context = ssl.create_default_context(cadata=self.root_certificates.decode("ascii")) + # A development certificate is self-signed and typically lacks the extensions strict mode demands. Trust is + # still limited to this one certificate and the host name is still checked. + context.verify_flags &= ~getattr(ssl, "VERIFY_X509_STRICT", 0) + return context + + +def is_loopback_host(host: str) -> bool: + """Return whether ``host`` is ``localhost`` or a loopback IP address.""" + if host.lower() == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +async def resolve_tls_trust( + options: ChronicleConnectionOptions, + *, + ca_certificates: bytes | None = None, + fetch_certificate: CertificateFetcher | None = None, +) -> TlsTrust | None: + """Resolve the trust for a connection, or ``None`` when the connection does not use TLS. + + Explicit ``ca_certificates`` always win. Otherwise the certificates are verified against the platform roots, + with one development exception: when ``skip_tls_validation`` is set (the connection string default) *and* the host + is a loopback address, the certificate the local kernel presents is trusted. The host name is still verified. For + any other host the option is ignored and the certificate must chain to a trusted root, so a remote kernel can + never be reached with relaxed validation. + """ + if not options.tls: + return None + if ca_certificates is not None: + return TlsTrust(ca_certificates) + if options.skip_tls_validation and is_loopback_host(options.host): + fetch = fetch_certificate or _fetch_certificate + pem = await asyncio.to_thread(fetch, options.host, options.port) + return TlsTrust(pem.encode("ascii")) + return TlsTrust(None) + + +def _fetch_certificate(host: str, port: int) -> str: + return ssl.get_server_certificate((host, port), timeout=10) diff --git a/src/cratis_chronicle/token_provider.py b/src/cratis_chronicle/token_provider.py new file mode 100644 index 0000000..dfb2dc6 --- /dev/null +++ b/src/cratis_chronicle/token_provider.py @@ -0,0 +1,184 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Async OAuth client-credentials token provider for Chronicle's ``/connect/token`` endpoint.""" + +from __future__ import annotations + +import asyncio +import json +import math +import time +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from .connection_string import ChronicleConnectionOptions +from .errors import TokenAuthorizationError, TokenRequestError, TokenResponseError +from .http_transport import HttpResponse, TokenTransport + +__all__ = ["DEFAULT_TOKEN_LIFETIME_SECONDS", "OAuthTokenProvider", "TOKEN_PATH"] + +TOKEN_PATH = "/connect/token" # noqa: S105 - an endpoint path, not a secret +DEFAULT_TOKEN_LIFETIME_SECONDS = 30.0 +"""The lifetime assumed when the endpoint omits ``expires_in``. Deliberately short so an unknown expiry is never trusted +for long.""" + +_MAX_REFRESH_MARGIN_SECONDS = 30.0 +_REDACTED = "****" + + +@dataclass(frozen=True, slots=True) +class _CachedToken: + value: str + refresh_at: float + + +class OAuthTokenProvider: + """Obtains, caches and refreshes the access token used for every new gRPC call. + + A token is reused until shortly before it expires. Concurrent callers share one in-flight request, and one caller + being cancelled does not cancel the request the others wait for. The client secret and the token never appear in + ``repr()``, ``str()`` or an exception message. + """ + + def __init__( + self, + options: ChronicleConnectionOptions, + transport: TokenTransport, + *, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._client_id = options.client_id + self._client_secret = options.client_secret + self._url = f"{'https' if options.tls else 'http'}://{_authority(options)}{TOKEN_PATH}" + self._transport = transport + self._clock = clock + self._token: _CachedToken | None = None + self._refresh: asyncio.Task[_CachedToken] | None = None + self._generation = 0 + self._in_flight: set[asyncio.Task[_CachedToken]] = set() + self._closed = False + + def __repr__(self) -> str: + return f"OAuthTokenProvider(url={self._url!r}, client_id={self._client_id!r})" + + async def get_token(self) -> str: + """Return a valid access token, requesting a new one when none is cached or it is about to expire. + + Raises: + TokenAuthorizationError: The endpoint rejected the credentials. + TokenResponseError: The endpoint returned an unusable response. + TokenRequestError: The request failed or the provider is closed. + """ + if self._closed: + raise TokenRequestError("The token provider is closed") + cached = self._token + if cached is not None and self._clock() < cached.refresh_at: + return cached.value + if self._refresh is None or self._refresh.done(): + self._refresh = asyncio.ensure_future(self._request_token(self._generation)) + self._in_flight.add(self._refresh) + self._refresh.add_done_callback(self._settled) + # Waiting through asyncio.wait, not awaiting the task: cancelling this caller must not cancel the request other + # callers share, and unlike asyncio.shield it leaves no wrapper future to report an unretrieved failure. + refresh = self._refresh + await asyncio.wait({refresh}) + return refresh.result().value + + def invalidate(self) -> None: + """Forget the cached token so the next call requests a new one. + + A request already in flight keeps serving the callers that are waiting on it, but its token is not cached and + later callers do not join it: it was issued before the token was declared invalid. + """ + self._generation += 1 + self._token = None + self._refresh = None + + async def aclose(self) -> None: + """Cancel any in-flight request and refuse further use. Safe to call more than once.""" + self._closed = True + self._token = None + self._refresh = None + pending = list(self._in_flight) + for refresh in pending: + refresh.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + def _settled(self, task: asyncio.Task[_CachedToken]) -> None: + """Forget a finished refresh and retrieve its outcome, so a failure nobody waits for is not left unretrieved.""" + self._in_flight.discard(task) + if not task.cancelled(): + task.exception() + + async def _request_token(self, generation: int) -> _CachedToken: + response = await self._transport.post_form( + self._url, + {"grant_type": "client_credentials", "client_id": self._client_id, "client_secret": self._client_secret}, + ) + token = self._interpret(response) + if not self._closed and generation == self._generation: + self._token = token + return token + + def _interpret(self, response: HttpResponse) -> _CachedToken: + if response.status in (400, 401, 403): + error, description = self._read_oauth_error(response.body) + raise TokenAuthorizationError(response.status, error, description) + if not 200 <= response.status < 300: + raise TokenRequestError(f"The token endpoint answered HTTP {response.status}", status=response.status) + + payload = self._parse_object(response.body) + access_token = payload.get("access_token") + if not isinstance(access_token, str) or not access_token.strip(): + raise TokenResponseError("The token response has no access_token") + token_type = payload.get("token_type") + if token_type is not None and (not isinstance(token_type, str) or token_type.lower() != "bearer"): + raise TokenResponseError("The token response has a token_type other than Bearer") + + lifetime = DEFAULT_TOKEN_LIFETIME_SECONDS + if "expires_in" in payload and payload["expires_in"] is not None: + expires_in = payload["expires_in"] + if ( + isinstance(expires_in, bool) + or not isinstance(expires_in, int | float) + or not math.isfinite(expires_in) + or expires_in <= 0 + ): + raise TokenResponseError("The token response has an invalid expires_in") + lifetime = float(expires_in) + margin = min(_MAX_REFRESH_MARGIN_SECONDS, lifetime / 2) + return _CachedToken(value=access_token, refresh_at=self._clock() + lifetime - margin) + + def _parse_object(self, body: bytes) -> dict[str, Any]: + try: + payload = json.loads(body) + except (ValueError, UnicodeDecodeError) as error: + raise TokenResponseError("The token response is not valid JSON") from error + if not isinstance(payload, dict): + raise TokenResponseError("The token response is not a JSON object") + return payload + + def _read_oauth_error(self, body: bytes) -> tuple[str | None, str | None]: + try: + payload = json.loads(body) + except (ValueError, UnicodeDecodeError): + return None, None + if not isinstance(payload, dict): + return None, None + return self._scrub(payload.get("error")), self._scrub(payload.get("error_description")) + + def _scrub(self, value: object) -> str | None: + if not isinstance(value, str): + return None + text = value[:200] + if self._client_secret: + text = text.replace(self._client_secret, _REDACTED) + return text + + +def _authority(options: ChronicleConnectionOptions) -> str: + host = f"[{options.host}]" if ":" in options.host else options.host + return f"{host}:{options.port}" diff --git a/src/cratis_chronicle/wire.py b/src/cratis_chronicle/wire.py new file mode 100644 index 0000000..8cd0605 --- /dev/null +++ b/src/cratis_chronicle/wire.py @@ -0,0 +1,103 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Conversion between Python values and Chronicle's wire values.""" + +from __future__ import annotations + +import dataclasses +import datetime +import enum +import json +import math +import uuid +from collections.abc import Mapping, Sequence +from typing import Any + +from cratis_chronicle_contracts.protobuf_net.bcl_pb2 import Guid + +__all__ = ["ConceptAs", "from_wire_guid", "serialize_content", "to_wire_guid", "to_wire_timestamp", "to_wire_value"] + + +@dataclasses.dataclass(frozen=True, slots=True) +class ConceptAs: + """Base for a strongly typed wrapper around a primitive. It serializes as its primitive ``value``.""" + + value: Any + + +def to_wire_guid(value: uuid.UUID) -> Guid: + """Convert a UUID to the protobuf-net ``Guid`` message: .NET byte order split into two little-endian halves.""" + raw = value.bytes_le + return Guid(lo=int.from_bytes(raw[:8], "little"), hi=int.from_bytes(raw[8:], "little")) + + +def from_wire_guid(value: Guid) -> uuid.UUID: + """Convert a protobuf-net ``Guid`` message back to a UUID.""" + return uuid.UUID(bytes_le=value.lo.to_bytes(8, "little") + value.hi.to_bytes(8, "little")) + + +def to_wire_timestamp(value: datetime.datetime) -> str: + """Convert a timezone-aware datetime to an ISO 8601 string with its UTC offset, as ``DateTimeOffset`` expects.""" + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("A datetime must be timezone-aware to be sent to Chronicle") + return value.isoformat() + + +def _to_wire_duration(value: datetime.timedelta) -> str: + """Convert a timedelta to the .NET constant ``TimeSpan`` form ``[-][d.]hh:mm:ss[.fffffff]``.""" + total_microseconds = (value.days * 86400 + value.seconds) * 1_000_000 + value.microseconds + sign = "-" if total_microseconds < 0 else "" + total_microseconds = abs(total_microseconds) + days, remainder = divmod(total_microseconds, 86_400 * 1_000_000) + hours, remainder = divmod(remainder, 3_600 * 1_000_000) + minutes, remainder = divmod(remainder, 60 * 1_000_000) + seconds, microseconds = divmod(remainder, 1_000_000) + text = f"{hours:02}:{minutes:02}:{seconds:02}" + if microseconds: + text += f".{microseconds * 10:07}" + return f"{sign}{days}.{text}" if days else f"{sign}{text}" + + +def to_wire_value(value: Any) -> Any: + """Convert a value to something ``json.dumps`` accepts, in the forms Chronicle expects. + + UUIDs become canonical strings, datetimes ISO 8601 with offset, dates and times ISO 8601, durations .NET + ``TimeSpan`` strings, enums and ``ConceptAs`` wrappers their primitive value. Dataclasses and mappings become + objects with their property names unchanged. Anything else is rejected rather than guessed. + """ + if value is None or isinstance(value, bool | int | str): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("NaN and infinity cannot be represented in an event") + return value + if isinstance(value, uuid.UUID): + return str(value) + if isinstance(value, datetime.datetime): + return to_wire_timestamp(value) + if isinstance(value, datetime.date | datetime.time): + return value.isoformat() + if isinstance(value, datetime.timedelta): + return _to_wire_duration(value) + if isinstance(value, ConceptAs): + return to_wire_value(value.value) + if isinstance(value, enum.Enum): + return to_wire_value(value.value) + if dataclasses.is_dataclass(value) and not isinstance(value, type): + return {field.name: to_wire_value(getattr(value, field.name)) for field in dataclasses.fields(value)} + if isinstance(value, Mapping): + if not all(isinstance(key, str) for key in value): + raise TypeError("Event property names must be strings") + return {key: to_wire_value(item) for key, item in value.items()} + if isinstance(value, Sequence) and not isinstance(value, bytes | bytearray): + return [to_wire_value(item) for item in value] + raise TypeError(f"A value of type {type(value).__name__} cannot be sent to Chronicle") + + +def serialize_content(content: Any) -> str: + """Serialize an event's content as the JSON object Chronicle stores. The content must be an object.""" + wire = to_wire_value(content) + if not isinstance(wire, dict): + raise TypeError("Event content must be a mapping or a dataclass") + return json.dumps(wire, separators=(",", ":"), allow_nan=False) diff --git a/tests/fake_kernel.py b/tests/fake_kernel.py new file mode 100644 index 0000000..6f9b208 --- /dev/null +++ b/tests/fake_kernel.py @@ -0,0 +1,116 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""An in-process gRPC server that stands in for the kernel services the client uses.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +import grpc +from cratis_chronicle_contracts import ( + eventstores_pb2, + eventstores_pb2_grpc, + eventtypes_pb2, + eventtypes_pb2_grpc, + namespaces_pb2, + namespaces_pb2_grpc, + sequences_pb2, + sequences_pb2_grpc, +) +from grpc import aio + +from cratis_chronicle import HttpResponse + + +@dataclass +class FakeKernel: + calls: list[str] = field(default_factory=list) + metadata: list[Mapping[str, str]] = field(default_factory=list) + appended: list[Any] = field(default_factory=list) + registered: list[Any] = field(default_factory=list) + append_errors: list[str] = field(default_factory=list) + namespace_failure: str | None = None + register_failure: str | None = None + next_sequence_number: int = 42 + port: int = 0 + _server: aio.Server | None = None + + def record(self, name: str, context: aio.ServicerContext) -> None: + self.calls.append(name) + self.metadata.append({key: value for key, value in context.invocation_metadata()}) + + async def start(self) -> None: + kernel = self + + class EventStores(eventstores_pb2_grpc.EventStoresServicer): + async def EnsureEventStore(self, request: Any, context: aio.ServicerContext) -> Any: + kernel.record(f"EnsureEventStore:{request.Name}", context) + return eventstores_pb2.CommandResult() + + class Namespaces(namespaces_pb2_grpc.NamespacesServicer): + async def EnsureNamespace(self, request: Any, context: aio.ServicerContext) -> Any: + kernel.record(f"EnsureNamespace:{request.EventStore}/{request.Namespace}", context) + result = namespaces_pb2.CommandResult() + if kernel.namespace_failure: + result.ExceptionMessages.append(kernel.namespace_failure) + return result + + class EventTypes(eventtypes_pb2_grpc.EventTypesServicer): + async def RegisterEventTypes(self, request: Any, context: aio.ServicerContext) -> Any: + kernel.record(f"RegisterEventTypes:{request.EventStore}", context) + kernel.registered.extend(request.Types) + result = eventtypes_pb2.CommandResult() + if kernel.register_failure: + result.ExceptionMessages.append(kernel.register_failure) + return result + + class EventSequences(sequences_pb2_grpc.EventSequencesServicer): + async def Append(self, request: Any, context: aio.ServicerContext) -> Any: + kernel.record("Append", context) + kernel.appended.append(request) + result = sequences_pb2.CommandResult_AppendResponse() + result.Response.SequenceNumber = kernel.next_sequence_number + result.Response.CorrelationId.CopyFrom(request.CorrelationId) + result.Response.Errors.extend(kernel.append_errors) + return result + + server = aio.server() + eventstores_pb2_grpc.add_EventStoresServicer_to_server(EventStores(), server) + namespaces_pb2_grpc.add_NamespacesServicer_to_server(Namespaces(), server) + eventtypes_pb2_grpc.add_EventTypesServicer_to_server(EventTypes(), server) + sequences_pb2_grpc.add_EventSequencesServicer_to_server(EventSequences(), server) + self.port = server.add_insecure_port("127.0.0.1:0") + self._server = server + await server.start() + + async def stop(self) -> None: + if self._server is not None: + await self._server.stop(None) + + def content_of_appended(self, index: int = 0) -> Any: + return json.loads(self.appended[index].Content) + + +class FakeTokenTransport: + """Hands out numbered tokens, so a test can see which token each call carried.""" + + def __init__(self, *, fail_with: BaseException | None = None, expires_in: float | None = 3600) -> None: + self.requests = 0 + self._fail_with = fail_with + self._expires_in = expires_in + + async def post_form(self, url: str, fields: Mapping[str, str]) -> HttpResponse: + self.requests += 1 + if self._fail_with is not None: + raise self._fail_with + body: dict[str, Any] = {"access_token": f"token-{self.requests}", "token_type": "Bearer"} + if self._expires_in is not None: + body["expires_in"] = self._expires_in + return HttpResponse(200, json.dumps(body).encode()) + + +__all__ = ["FakeKernel", "FakeTokenTransport", "grpc"] diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..8207230 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,208 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import asyncio +import datetime +import uuid +from collections.abc import Awaitable, Callable +from typing import Any + +import grpc +import pytest +from fake_kernel import FakeKernel, FakeTokenTransport + +from cratis_chronicle import ( + AppendFailedError, + ChronicleChannel, + ChronicleClient, + ChronicleConnectionOptions, + CommandFailedError, + EventTypeDefinition, + TokenRequestError, +) + +BOOK_ADDED = EventTypeDefinition( + id="book-added", schema={"type": "object", "properties": {"title": {"type": "string"}}} +) + + +def with_kernel( + scenario: Callable[[FakeKernel, ChronicleClient, FakeTokenTransport], Awaitable[Any]], + transport: FakeTokenTransport | None = None, + kernel: FakeKernel | None = None, +) -> Any: + async def run() -> Any: + fake = kernel or FakeKernel() + tokens = transport or FakeTokenTransport() + await fake.start() + try: + options = ChronicleConnectionOptions( + host="127.0.0.1", client_id="id", client_secret="secret", port=fake.port, tls=False + ) + channel = await ChronicleChannel.open(options, token_transport=tokens) + async with ChronicleClient(channel) as client: + return await scenario(fake, client, tokens) + finally: + await fake.stop() + + return asyncio.run(run()) + + +def test_establishes_state_in_order_registers_a_schema_and_appends_to_the_event_log() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> Any: + store = await client.ensure_event_store("books") + namespace = await store.ensure_namespace() + await store.register_event_type(BOOK_ADDED) + source = str(uuid.uuid4()) + return source, await namespace.event_log.append(source, BOOK_ADDED, {"title": "Dune"}) + + kernel = FakeKernel() + source, result = with_kernel(scenario, kernel=kernel) + + assert kernel.calls == [ + "EnsureEventStore:books", + "EnsureNamespace:books/Default", + "RegisterEventTypes:books", + "Append", + ] + registration = kernel.registered[0] + assert (registration.Type.Id, registration.Type.Generation) == ("book-added", 1) + assert '"properties"' in registration.Schema + request = kernel.appended[0] + assert (request.EventStore, request.Namespace, request.EventSequenceId) == ("books", "Default", "event-log") + assert request.EventSourceId == source + assert (request.EventType.Id, request.EventType.Generation) == ("book-added", 1) + assert kernel.content_of_appended() == {"title": "Dune"} + assert request.ConcurrencyScope.SequenceNumber == 2**64 - 1 + assert len(request.Causation) == 1 + assert result.sequence_number == 42 + assert result.correlation_id.int != 0 + + +def test_every_call_carries_the_current_bearer_token() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + store = await client.ensure_event_store("books") + await store.ensure_namespace() + + kernel, transport = FakeKernel(), FakeTokenTransport() + with_kernel(scenario, transport, kernel) + + assert [entry["authorization"] for entry in kernel.metadata] == ["Bearer token-1", "Bearer token-1"] + assert transport.requests == 1 + + +def test_a_refreshed_token_is_used_for_the_next_new_call() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + await client.ensure_event_store("a") + await asyncio.sleep(0.05) + await client.ensure_event_store("b") + + kernel = FakeKernel() + with_kernel(scenario, FakeTokenTransport(expires_in=0.05), kernel) + + assert [entry["authorization"] for entry in kernel.metadata] == ["Bearer token-1", "Bearer token-2"] + + +def test_a_token_failure_surfaces_as_a_token_error_and_nothing_reaches_the_kernel() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + await client.ensure_event_store("books") + + kernel = FakeKernel() + transport = FakeTokenTransport(fail_with=TokenRequestError("endpoint down")) + + with pytest.raises(TokenRequestError, match="endpoint down"): + with_kernel(scenario, transport, kernel) + + assert kernel.calls == [] + + +def test_a_failed_command_result_raises_with_the_reason() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + store = await client.ensure_event_store("books") + await store.ensure_namespace() + + kernel = FakeKernel(namespace_failure="boom") + + with pytest.raises(CommandFailedError, match="boom") as raised: + with_kernel(scenario, kernel=kernel) + + assert raised.value.reasons == ["boom"] + + +def test_kernel_append_errors_raise_append_failed() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + namespace = await (await client.ensure_event_store("books")).ensure_namespace() + await namespace.event_log.append("src", BOOK_ADDED, {"title": "Dune"}) + + with pytest.raises(AppendFailedError, match="schema mismatch"): + with_kernel(scenario, kernel=FakeKernel(append_errors=["schema mismatch"])) + + +def test_invalid_inputs_are_rejected_before_anything_is_sent() -> None: + async def scenario(kernel: FakeKernel, client: ChronicleClient, _: FakeTokenTransport) -> None: + namespace = await (await client.ensure_event_store("books")).ensure_namespace() + with pytest.raises(ValueError, match="event source id"): + await namespace.event_log.append(" ", BOOK_ADDED, {}) + with pytest.raises(ValueError, match="timezone-aware"): + await namespace.event_log.append("a", BOOK_ADDED, {}, occurred=datetime.datetime(2026, 1, 1)) + with pytest.raises(TypeError): + await namespace.event_log.append("a", BOOK_ADDED, [1]) + + kernel = FakeKernel() + with_kernel(scenario, kernel=kernel) + + assert "Append" not in kernel.calls + + +def test_event_type_definitions_require_an_id_a_generation_and_a_non_empty_schema() -> None: + with pytest.raises(ValueError, match="non-empty JSON schema"): + EventTypeDefinition(id="x", schema={}) + with pytest.raises(ValueError, match="id"): + EventTypeDefinition(id=" ", schema={"type": "object"}) + with pytest.raises(ValueError, match="generation"): + EventTypeDefinition(id="x", schema={"type": "object"}, generation=0) + + +def test_closing_is_deterministic_and_idempotent() -> None: + async def run() -> None: + kernel = FakeKernel() + await kernel.start() + try: + options = ChronicleConnectionOptions( + host="127.0.0.1", client_id="id", client_secret="secret", port=kernel.port, tls=False + ) + client = ChronicleClient(await ChronicleChannel.open(options, token_transport=FakeTokenTransport())) + await client.ensure_event_store("books") + await client.aclose() + await client.aclose() + with pytest.raises((grpc.aio.UsageError, TokenRequestError)): + await client.ensure_event_store("books") + finally: + await kernel.stop() + + asyncio.run(run()) + + +def test_connect_accepts_a_connection_string() -> None: + async def run() -> None: + client = await ChronicleClient.connect("chronicle://id:secret@localhost:1/?skipTlsValidation=false") + await client.aclose() + + asyncio.run(run()) + + +def test_credentials_are_not_sent_without_tls_to_a_remote_host() -> None: + from cratis_chronicle import ChronicleChannel, ChronicleConnectionOptions, InsecureTransportError + + options = ChronicleConnectionOptions(host="kernel.example.com", client_id="id", client_secret="secret", tls=False) + with pytest.raises(InsecureTransportError): + asyncio.run(ChronicleChannel.open(options)) + + +def test_a_failed_event_type_registration_is_reported_instead_of_ignored() -> None: + async def scenario(_: FakeKernel, client: ChronicleClient, __: FakeTokenTransport) -> None: + store = await client.ensure_event_store("books") + await store.register_event_type(BOOK_ADDED) + + with pytest.raises(CommandFailedError, match="schema rejected"): + with_kernel(scenario, kernel=FakeKernel(register_failure="schema rejected")) diff --git a/tests/test_contract_dependency.py b/tests/test_contract_dependency.py index 6cf2e04..5fd78bf 100644 --- a/tests/test_contract_dependency.py +++ b/tests/test_contract_dependency.py @@ -1,9 +1,17 @@ # Copyright (c) Cratis. All rights reserved. # Licensed under the MIT license. See LICENSE file in the project root for full license information. -from cratis_chronicle_contracts.events_pb2 import EventType -from cratis_chronicle_contracts.events_pb2_grpc import EventTypesStub +from importlib.metadata import version + from cratis_chronicle_contracts.protobuf_net.bcl_pb2 import Guid +from cratis_chronicle_contracts.sequences_pb2 import AppendRequest, EventType +from cratis_chronicle_contracts.sequences_pb2_grpc import EventSequencesStub + +from cratis_chronicle import _event_type_contracts as event_types + + +def test_generated_contract_dependency_is_the_pinned_release() -> None: + assert version("cratis-chronicle-contracts") == "19.31.3" def test_generated_contract_dependency_is_available() -> None: @@ -14,4 +22,7 @@ def test_generated_contract_dependency_is_available() -> None: assert event_type.Generation == 1 assert guid.lo == 1 assert guid.hi == 2 - assert EventTypesStub is not None + assert AppendRequest(EventType=event_type).EventType.Id == "example" + assert EventSequencesStub is not None + assert event_types.services.EventTypesStub is not None + assert event_types.messages.RegisterEventTypesRequest is not None diff --git a/tests/test_http_transport.py b/tests/test_http_transport.py new file mode 100644 index 0000000..a2923f3 --- /dev/null +++ b/tests/test_http_transport.py @@ -0,0 +1,253 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import asyncio +import shutil +import ssl +import subprocess +from collections.abc import Awaitable, Callable +from pathlib import Path + +import pytest + +from cratis_chronicle import StreamFormTransport, TlsTrust, TokenRequestError +from cratis_chronicle.http_transport import _decode_chunked, _parse_response + +Handler = Callable[[asyncio.StreamReader, asyncio.StreamWriter], Awaitable[None]] + + +async def serve(handler: Handler, context: ssl.SSLContext | None = None) -> tuple[asyncio.Server, int]: + server = await asyncio.start_server(handler, "127.0.0.1", 0, ssl=context) + return server, server.sockets[0].getsockname()[1] + + +async def read_request(reader: asyncio.StreamReader) -> tuple[str, bytes]: + head = await reader.readuntil(b"\r\n\r\n") + length = next( + int(line.split(b":")[1]) for line in head.split(b"\r\n") if line.lower().startswith(b"content-length") + ) + return head.decode(), await reader.readexactly(length) + + +def test_posts_a_form_and_reads_a_content_length_response() -> None: + seen: list[tuple[str, bytes]] = [] + + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + seen.append(await read_request(reader)) + writer.write(b'HTTP/1.1 200 OK\r\nContent-Length: 7\r\n\r\n{"a":1}') + await writer.drain() + writer.close() + + async def scenario() -> tuple[int, bytes]: + server, port = await serve(handler) + async with server: + response = await StreamFormTransport(None).post_form( + f"http://127.0.0.1:{port}/connect/token", {"client_secret": "a b&c", "grant_type": "x"} + ) + return response.status, response.body + + assert asyncio.run(scenario()) == (200, b'{"a":1}') + head, body = seen[0] + assert head.startswith("POST /connect/token HTTP/1.1") + assert "content-type: application/x-www-form-urlencoded" in head.lower() + assert body == b"client_secret=a+b%26c&grant_type=x" + + +def test_reads_a_chunked_response() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b"HTTP/1.1 401 Unauthorized\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nabc\r\n2\r\nde\r\n0\r\n\r\n") + await writer.drain() + writer.close() + + async def scenario() -> tuple[int, bytes]: + server, port = await serve(handler) + async with server: + response = await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {}) + return response.status, response.body + + assert asyncio.run(scenario()) == (401, b"abcde") + + +def test_a_redirect_is_returned_not_followed() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b"HTTP/1.1 302 Found\r\nLocation: http://elsewhere.invalid/\r\nContent-Length: 0\r\n\r\n") + await writer.drain() + writer.close() + + async def scenario() -> int: + server, port = await serve(handler) + async with server: + return (await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {})).status + + assert asyncio.run(scenario()) == 302 + + +def test_a_timeout_is_a_request_error() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + await asyncio.sleep(1) + writer.close() + + async def scenario() -> None: + server, port = await serve(handler) + async with server: + await StreamFormTransport(None, timeout=0.2).post_form(f"http://127.0.0.1:{port}/x", {}) + + with pytest.raises(TokenRequestError, match="timed out"): + asyncio.run(scenario()) + + +def test_a_refused_connection_is_a_request_error_without_secrets() -> None: + async def scenario() -> None: + server, port = await serve(lambda r, w: asyncio.sleep(0)) # type: ignore[arg-type, return-value] + server.close() + await server.wait_closed() + await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {"client_secret": "hunter2"}) + + with pytest.raises(TokenRequestError) as raised: + asyncio.run(scenario()) + + assert "hunter2" not in str(raised.value) + + +def test_a_garbled_response_is_a_request_error() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b"garbage") + await writer.drain() + writer.close() + + async def scenario() -> None: + server, port = await serve(handler) + async with server: + await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {}) + + with pytest.raises(TokenRequestError): + asyncio.run(scenario()) + + +def test_chunk_and_status_parsers_reject_malformed_input() -> None: + with pytest.raises(ValueError): + _decode_chunked(b"zz\r\nabc") + with pytest.raises(ValueError): + _decode_chunked(b"5\r\nab") + with pytest.raises(ValueError): + _parse_response(b"HTTP/1.1 abc\r\n\r\n") + + +@pytest.fixture +def self_signed(tmp_path: Path) -> tuple[Path, Path]: + openssl = shutil.which("openssl") + if openssl is None: + pytest.skip("openssl is not available") + cert, key = tmp_path / "cert.pem", tmp_path / "key.pem" + subprocess.run( # noqa: S603 + [ + openssl, + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + str(key), + "-out", + str(cert), + "-days", + "1", + "-subj", + "/CN=localhost", + "-addext", + "subjectAltName=DNS:localhost,IP:127.0.0.1", + ], + check=True, + capture_output=True, + ) + return cert, key + + +def tls_scenario(self_signed: tuple[Path, Path], transport: StreamFormTransport) -> int: + cert, key = self_signed + server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + server_context.load_cert_chain(cert, key) + + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}") + await writer.drain() + writer.close() + + async def scenario() -> int: + server, port = await serve(handler, server_context) + async with server: + return (await transport.post_form(f"https://localhost:{port}/x", {})).status + + return asyncio.run(scenario()) + + +def test_an_untrusted_self_signed_certificate_is_rejected_by_default(self_signed: tuple[Path, Path]) -> None: + transport = StreamFormTransport(TlsTrust(None).ssl_context()) + + with pytest.raises(TokenRequestError, match="TLS handshake"): + tls_scenario(self_signed, transport) + + +def test_explicitly_trusting_the_certificate_allows_the_connection(self_signed: tuple[Path, Path]) -> None: + trust = TlsTrust(self_signed[0].read_bytes()) + + assert tls_scenario(self_signed, StreamFormTransport(trust.ssl_context())) == 200 + + +def test_trusting_one_certificate_still_verifies_the_host_name(self_signed: tuple[Path, Path]) -> None: + cert, key = self_signed + context = TlsTrust(cert.read_bytes()).ssl_context() + server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + server_context.load_cert_chain(cert, key) + + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + writer.close() + + async def scenario() -> None: + server, port = await serve(handler, server_context) + async with server: + # 127.0.0.1 is in the certificate, but a different name is not. + await StreamFormTransport(context).post_form(f"https://localhost.localdomain:{port}/x", {}) + + with pytest.raises(TokenRequestError): + asyncio.run(scenario()) + + +def test_reads_a_response_that_arrives_in_several_segments() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 7\r\n\r\n") + await writer.drain() + await asyncio.sleep(0.05) + writer.write(b'{"a":1}') + await writer.drain() + writer.close() + + async def scenario() -> bytes: + server, port = await serve(handler) + async with server: + return (await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {})).body + + assert asyncio.run(scenario()) == b'{"a":1}' + + +def test_a_body_shorter_than_its_content_length_is_rejected() -> None: + async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await read_request(reader) + writer.write(b'HTTP/1.1 200 OK\r\nContent-Length: 50\r\n\r\n{"access_tok') + await writer.drain() + writer.close() + + async def scenario() -> None: + server, port = await serve(handler) + async with server: + await StreamFormTransport(None).post_form(f"http://127.0.0.1:{port}/x", {}) + + with pytest.raises(TokenRequestError): + asyncio.run(scenario()) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..fdd09ac --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,94 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +"""Opt-in checks against a real kernel. Skipped unless CHRONICLE_INTEGRATION_URL names one. + +Set it to the connection string of a *development* kernel you started for this purpose, for example +``CHRONICLE_INTEGRATION_URL=chronicle://localhost:19300``. See Documentation/client-development-guide.md. +""" + +import asyncio +import os +import uuid +from urllib.parse import urlsplit + +import pytest + +from cratis_chronicle import ( + AppendFailedError, + ChronicleClient, + EventTypeDefinition, + TokenAuthorizationError, + TokenError, +) + +URL = os.environ.get("CHRONICLE_INTEGRATION_URL") +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif(URL is None, reason="CHRONICLE_INTEGRATION_URL is not set"), +] + +EVENT_TYPE = EventTypeDefinition( + id="python-integration.book-added", + schema={"type": "object", "properties": {"title": {"type": "string"}}, "required": ["title"]}, +) + + +def endpoint() -> str: + assert URL is not None + parts = urlsplit(URL) + return f"{parts.hostname}:{parts.port}" + + +def test_authenticates_establishes_state_registers_and_appends_to_the_event_log() -> None: + async def scenario() -> tuple[int, int]: + assert URL is not None + async with await ChronicleClient.connect(URL) as client: + store = await client.ensure_event_store("python-integration") + namespace = await store.ensure_namespace("Default") + await store.register_event_type(EVENT_TYPE) + source = str(uuid.uuid4()) + first = await namespace.event_log.append(source, EVENT_TYPE, {"title": "first"}) + second = await namespace.event_log.append(source, EVENT_TYPE, {"title": "second"}) + return first.sequence_number, second.sequence_number + + first, second = asyncio.run(scenario()) + + assert first >= 0 + assert second == first + 1 + + +def test_content_that_violates_the_registered_schema_is_not_appended() -> None: + async def scenario() -> None: + assert URL is not None + async with await ChronicleClient.connect(URL) as client: + store = await client.ensure_event_store("python-integration") + namespace = await store.ensure_namespace("Default") + await store.register_event_type(EVENT_TYPE) + await namespace.event_log.append(str(uuid.uuid4()), EVENT_TYPE, {"unexpected": 1}) + + with pytest.raises(AppendFailedError, match="title"): + asyncio.run(scenario()) + + +def test_wrong_credentials_fail_with_a_token_authorization_error_that_does_not_leak_the_secret() -> None: + secret = "definitely-the-wrong-secret" # noqa: S105 + + async def scenario() -> None: + async with await ChronicleClient.connect(f"chronicle://chronicle-dev-client:{secret}@{endpoint()}") as client: + await client.ensure_event_store("python-integration") + + with pytest.raises(TokenAuthorizationError) as raised: + asyncio.run(scenario()) + + assert secret not in str(raised.value) + assert secret not in repr(raised.value) + + +def test_a_self_signed_kernel_certificate_is_rejected_when_validation_is_required() -> None: + async def scenario() -> None: + async with await ChronicleClient.connect(f"chronicle://{endpoint()}/?skipTlsValidation=false") as client: + await client.ensure_event_store("python-integration") + + with pytest.raises(TokenError, match="TLS handshake"): + asyncio.run(scenario()) diff --git a/tests/test_tls.py b/tests/test_tls.py new file mode 100644 index 0000000..5b2eead --- /dev/null +++ b/tests/test_tls.py @@ -0,0 +1,67 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import asyncio + +import pytest + +from cratis_chronicle import ChronicleConnectionOptions, is_loopback_host, resolve_tls_trust + + +def options(host: str, **overrides: object) -> ChronicleConnectionOptions: + return ChronicleConnectionOptions(host=host, client_id="id", client_secret="secret", **overrides) # type: ignore[arg-type] + + +def resolve(opts: ChronicleConnectionOptions, ca: bytes | None = None) -> tuple[object, list[tuple[str, int]]]: + fetched: list[tuple[str, int]] = [] + + def fetch(host: str, port: int) -> str: + fetched.append((host, port)) + return "-----BEGIN CERTIFICATE-----\nabc\n-----END CERTIFICATE-----\n" + + return asyncio.run(resolve_tls_trust(opts, ca_certificates=ca, fetch_certificate=fetch)), fetched + + +@pytest.mark.parametrize("host", ["localhost", "LOCALHOST", "127.0.0.1", "127.5.5.5", "::1"]) +def test_loopback_hosts_are_recognized(host: str) -> None: + assert is_loopback_host(host) + + +@pytest.mark.parametrize("host", ["kernel.example", "10.0.0.5", "localhost.example.com", "2001:db8::1"]) +def test_other_hosts_are_not_loopback(host: str) -> None: + assert not is_loopback_host(host) + + +def test_a_loopback_development_kernel_certificate_is_trusted_when_skip_validation_is_set() -> None: + trust, fetched = resolve(options("localhost", port=19300)) + + assert fetched == [("localhost", 19300)] + assert trust.root_certificates is not None # type: ignore[attr-defined] + + +def test_a_remote_host_never_gets_relaxed_validation_even_when_skip_validation_is_set() -> None: + trust, fetched = resolve(options("kernel.example")) + + assert fetched == [] + assert trust.root_certificates is None # type: ignore[attr-defined] + + +def test_loopback_verifies_normally_when_skip_validation_is_off() -> None: + trust, fetched = resolve(options("localhost", skip_tls_validation=False)) + + assert fetched == [] + assert trust.root_certificates is None # type: ignore[attr-defined] + + +def test_explicit_ca_certificates_win_and_nothing_is_fetched() -> None: + trust, fetched = resolve(options("localhost"), ca=b"PEM") + + assert fetched == [] + assert trust.root_certificates == b"PEM" # type: ignore[attr-defined] + + +def test_no_trust_is_resolved_without_tls() -> None: + trust, fetched = resolve(options("localhost", tls=False)) + + assert trust is None + assert fetched == [] diff --git a/tests/test_token_provider.py b/tests/test_token_provider.py new file mode 100644 index 0000000..8bc65df --- /dev/null +++ b/tests/test_token_provider.py @@ -0,0 +1,360 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import asyncio +import gc +import json +import warnings +from collections.abc import Mapping +from typing import Any + +import pytest + +from cratis_chronicle import ( + ChronicleConnectionOptions, + HttpResponse, + OAuthTokenProvider, + TokenAuthorizationError, + TokenRequestError, + TokenResponseError, +) +from cratis_chronicle.token_provider import DEFAULT_TOKEN_LIFETIME_SECONDS + +SECRET = "s3cr3t-value" # noqa: S105 +OPTIONS = ChronicleConnectionOptions(host="kernel.example", client_id="the-client", client_secret=SECRET, port=35001) + + +def ok(token: str = "tok-1", **extra: Any) -> HttpResponse: + return HttpResponse(200, json.dumps({"access_token": token, "token_type": "Bearer", **extra}).encode()) + + +class FakeTransport: + def __init__(self, *responses: HttpResponse | BaseException) -> None: + self.responses = list(responses) + self.calls: list[tuple[str, Mapping[str, str]]] = [] + self.gate: asyncio.Event | None = None + + async def post_form(self, url: str, fields: Mapping[str, str]) -> HttpResponse: + self.calls.append((url, dict(fields))) + if self.gate is not None: + await self.gate.wait() + response = self.responses.pop(0) if len(self.responses) > 1 else self.responses[0] + if isinstance(response, BaseException): + raise response + return response + + +class Clock: + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + +def provider(transport: FakeTransport, clock: Clock | None = None) -> OAuthTokenProvider: + return OAuthTokenProvider(OPTIONS, transport, clock=clock or Clock()) + + +def test_requests_the_token_with_the_client_credentials_form() -> None: + transport = FakeTransport(ok(expires_in=3600)) + + token = asyncio.run(provider(transport).get_token()) + + assert token == "tok-1" + assert transport.calls == [ + ( + "https://kernel.example:35001/connect/token", + {"grant_type": "client_credentials", "client_id": "the-client", "client_secret": SECRET}, + ) + ] + + +def test_uses_http_when_tls_is_off_and_brackets_ipv6() -> None: + options = ChronicleConnectionOptions(host="::1", client_id="a", client_secret="b", tls=False) + transport = FakeTransport(ok()) + + asyncio.run(OAuthTokenProvider(options, transport).get_token()) + + assert transport.calls[0][0] == "http://[::1]:35000/connect/token" + + +def test_caches_the_token_until_shortly_before_it_expires() -> None: + clock, transport = Clock(), FakeTransport(ok("a", expires_in=100), ok("b", expires_in=100)) + tokens = provider(transport, clock) + + async def scenario() -> list[str]: + seen = [await tokens.get_token()] + clock.now += 69 # lifetime 100 -> refresh margin 30 -> still valid at 70 + seen.append(await tokens.get_token()) + clock.now += 2 # past refresh point + seen.append(await tokens.get_token()) + return seen + + assert asyncio.run(scenario()) == ["a", "a", "b"] + assert len(transport.calls) == 2 + + +def test_an_absent_expiry_is_cached_only_for_a_short_conservative_time() -> None: + clock, transport = Clock(), FakeTransport(ok("a"), ok("b")) + tokens = provider(transport, clock) + + async def scenario() -> list[str]: + seen = [await tokens.get_token()] + clock.now += DEFAULT_TOKEN_LIFETIME_SECONDS / 2 - 1 + seen.append(await tokens.get_token()) + clock.now += 2 + seen.append(await tokens.get_token()) + return seen + + assert asyncio.run(scenario()) == ["a", "a", "b"] + + +def test_short_lifetimes_refresh_at_half_their_life() -> None: + clock, transport = Clock(), FakeTransport(ok("a", expires_in=10), ok("b", expires_in=10)) + tokens = provider(transport, clock) + + async def scenario() -> list[str]: + first = await tokens.get_token() + clock.now += 6 + return [first, await tokens.get_token()] + + assert asyncio.run(scenario()) == ["a", "b"] + + +def test_concurrent_callers_share_one_request() -> None: + transport = FakeTransport(ok(expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> list[str]: + transport.gate = asyncio.Event() + callers = [asyncio.ensure_future(tokens.get_token()) for _ in range(10)] + await asyncio.sleep(0.01) + transport.gate.set() + return await asyncio.gather(*callers) + + assert asyncio.run(scenario()) == ["tok-1"] * 10 + assert len(transport.calls) == 1 + + +def test_cancelling_one_caller_does_not_cancel_the_shared_request() -> None: + transport = FakeTransport(ok(expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> str: + transport.gate = asyncio.Event() + cancelled = asyncio.ensure_future(tokens.get_token()) + survivor = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + cancelled.cancel() + transport.gate.set() + with pytest.raises(asyncio.CancelledError): + await cancelled + return await survivor + + assert asyncio.run(scenario()) == "tok-1" + assert len(transport.calls) == 1 + + +def test_closing_cancels_the_in_flight_request() -> None: + transport = FakeTransport(ok(expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> str: + transport.gate = asyncio.Event() + caller = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + await tokens.aclose() + with pytest.raises(asyncio.CancelledError): + await caller + return "closed" + + assert asyncio.run(scenario()) == "closed" + + +def test_close_is_idempotent_and_stops_further_use() -> None: + tokens = provider(FakeTransport(ok())) + + async def scenario() -> None: + await tokens.aclose() + await tokens.aclose() + with pytest.raises(TokenRequestError): + await tokens.get_token() + + asyncio.run(scenario()) + + +def test_invalidate_forces_a_new_request() -> None: + transport = FakeTransport(ok("a", expires_in=3600), ok("b", expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> list[str]: + first = await tokens.get_token() + tokens.invalidate() + return [first, await tokens.get_token()] + + assert asyncio.run(scenario()) == ["a", "b"] + + +@pytest.mark.parametrize( + "body", + [ + b"not json", + b"[]", + b"{}", + b'{"access_token": ""}', + b'{"access_token": 5}', + b'{"access_token": "t", "token_type": "mac"}', + b'{"access_token": "t", "expires_in": "soon"}', + b'{"access_token": "t", "expires_in": 0}', + b'{"access_token": "t", "expires_in": -5}', + b'{"access_token": "t", "expires_in": true}', + ], +) +def test_malformed_responses_are_rejected(body: bytes) -> None: + tokens = provider(FakeTransport(HttpResponse(200, body))) + + with pytest.raises(TokenResponseError): + asyncio.run(tokens.get_token()) + + +def test_a_null_expires_in_is_treated_as_absent() -> None: + tokens = provider(FakeTransport(HttpResponse(200, b'{"access_token": "t", "expires_in": null}'))) + + assert asyncio.run(tokens.get_token()) == "t" + + +def test_authorization_failure_is_parsed_and_typed() -> None: + body = json.dumps({"error": "invalid_client", "error_description": "bad client"}).encode() + tokens = provider(FakeTransport(HttpResponse(401, body))) + + with pytest.raises(TokenAuthorizationError) as raised: + asyncio.run(tokens.get_token()) + + assert raised.value.status == 401 + assert raised.value.error == "invalid_client" + assert raised.value.description == "bad client" + assert "invalid_client" in str(raised.value) + + +def test_a_failed_request_is_not_cached() -> None: + transport = FakeTransport(HttpResponse(401, b"{}"), ok()) + tokens = provider(transport) + + async def scenario() -> str: + with pytest.raises(TokenAuthorizationError): + await tokens.get_token() + return await tokens.get_token() + + assert asyncio.run(scenario()) == "tok-1" + + +def test_an_unexpected_status_is_a_request_error_with_the_status() -> None: + tokens = provider(FakeTransport(HttpResponse(503, b"down"))) + + with pytest.raises(TokenRequestError) as raised: + asyncio.run(tokens.get_token()) + + assert raised.value.status == 503 + + +def test_secrets_never_leak_into_repr_or_errors() -> None: + echoing = json.dumps({"error": "invalid_client", "error_description": f"wrong {SECRET}"}).encode() + tokens = provider(FakeTransport(HttpResponse(401, echoing))) + + with pytest.raises(TokenAuthorizationError) as raised: + asyncio.run(tokens.get_token()) + + assert SECRET not in repr(tokens) + assert SECRET not in str(tokens) + assert SECRET not in str(raised.value) + assert SECRET not in repr(raised.value.description) + assert "****" in str(raised.value) + + +def test_the_access_token_never_appears_in_repr_or_response_errors() -> None: + tokens = provider(FakeTransport(ok("super-token", expires_in=3600))) + asyncio.run(tokens.get_token()) + + assert "super-token" not in repr(tokens) + assert "super-token" not in str(tokens) + + +def test_invalidating_during_a_refresh_does_not_cache_the_stale_token() -> None: + transport = FakeTransport(ok("stale", expires_in=3600), ok("fresh", expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> list[str]: + transport.gate = asyncio.Event() + in_flight = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + tokens.invalidate() + transport.gate.set() + waiter = await in_flight + transport.gate = None + return [waiter, await tokens.get_token(), await tokens.get_token()] + + assert asyncio.run(scenario()) == ["stale", "fresh", "fresh"] + assert len(transport.calls) == 2 + + +def test_a_caller_after_invalidation_does_not_join_the_stale_refresh() -> None: + transport = FakeTransport(ok("stale", expires_in=3600), ok("fresh", expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> str: + transport.gate = asyncio.Event() + stale = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + tokens.invalidate() + later = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + transport.gate.set() + await stale + return await later + + assert asyncio.run(scenario()) == "fresh" + + +def test_a_failed_shared_refresh_whose_waiters_all_cancelled_leaves_no_unretrieved_exception() -> None: + transport = FakeTransport(TokenRequestError("unreachable")) + tokens = provider(transport) + unhandled: list[dict[str, Any]] = [] + + async def scenario() -> None: + asyncio.get_running_loop().set_exception_handler(lambda _, context: unhandled.append(context)) + transport.gate = asyncio.Event() + first = asyncio.ensure_future(tokens.get_token()) + second = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + first.cancel() + second.cancel() + await asyncio.gather(first, second, return_exceptions=True) + transport.gate.set() + await asyncio.sleep(0.01) + gc.collect() + await asyncio.sleep(0) + + with warnings.catch_warnings(): + warnings.simplefilter("error") + asyncio.run(scenario()) + gc.collect() + + assert unhandled == [] + assert len(transport.calls) == 1 + + +def test_closing_after_invalidation_still_cancels_the_detached_refresh() -> None: + transport = FakeTransport(ok(expires_in=3600)) + tokens = provider(transport) + + async def scenario() -> None: + transport.gate = asyncio.Event() + caller = asyncio.ensure_future(tokens.get_token()) + await asyncio.sleep(0.01) + tokens.invalidate() + await tokens.aclose() + with pytest.raises(asyncio.CancelledError): + await caller + + asyncio.run(scenario()) diff --git a/tests/test_wire.py b/tests/test_wire.py new file mode 100644 index 0000000..453df82 --- /dev/null +++ b/tests/test_wire.py @@ -0,0 +1,119 @@ +# Copyright (c) Cratis. All rights reserved. +# Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import dataclasses +import datetime +import enum +import json +import uuid + +import pytest + +from cratis_chronicle import ConceptAs +from cratis_chronicle.wire import from_wire_guid, serialize_content, to_wire_guid, to_wire_value + + +def test_guid_round_trips_in_dotnet_byte_order() -> None: + value = uuid.UUID("00112233-4455-6677-8899-aabbccddeeff") + + wire = to_wire_guid(value) + + assert wire.lo == int.from_bytes(bytes.fromhex("3322110055447766"), "little") + assert wire.hi == int.from_bytes(bytes.fromhex("8899aabbccddeeff"), "little") + assert from_wire_guid(wire) == value + + +def test_random_guids_round_trip() -> None: + for _ in range(50): + value = uuid.uuid4() + assert from_wire_guid(to_wire_guid(value)) == value + + +class Color(enum.Enum): + RED = "red" + + +class Id(ConceptAs): + pass + + +@dataclasses.dataclass +class Address: + city: str + + +@dataclasses.dataclass +class Person: + id: uuid.UUID + ident: Id + when: datetime.datetime + born: datetime.date + at: datetime.time + took: datetime.timedelta + color: Color + address: Address + tags: list[str] + nickname: str | None = None + + +def test_values_are_converted_to_chronicle_forms_with_property_names_untouched() -> None: + person = Person( + id=uuid.UUID("12345678-1234-5678-1234-567812345678"), + ident=Id("abc"), + when=datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.timezone(datetime.timedelta(hours=2))), + born=datetime.date(1990, 5, 6), + at=datetime.time(7, 8, 9), + took=datetime.timedelta(days=1, hours=2, minutes=3, seconds=4, microseconds=500000), + color=Color.RED, + address=Address("Oslo"), + tags=["a", "b"], + ) + + assert json.loads(serialize_content(person)) == { + "id": "12345678-1234-5678-1234-567812345678", + "ident": "abc", + "when": "2026-01-02T03:04:05+02:00", + "born": "1990-05-06", + "at": "07:08:09", + "took": "1.02:03:04.5000000", + "color": "red", + "address": {"city": "Oslo"}, + "tags": ["a", "b"], + "nickname": None, + } + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (datetime.timedelta(0), "00:00:00"), + (datetime.timedelta(seconds=90), "00:01:30"), + (datetime.timedelta(days=3), "3.00:00:00"), + (datetime.timedelta(seconds=-90), "-00:01:30"), + (datetime.timedelta(microseconds=1), "00:00:00.0000010"), + ], +) +def test_durations_use_the_dotnet_constant_form(value: datetime.timedelta, expected: str) -> None: + assert to_wire_value(value) == expected + + +def test_naive_datetimes_are_rejected() -> None: + with pytest.raises(ValueError, match="timezone-aware"): + to_wire_value(datetime.datetime(2026, 1, 1)) + + +@pytest.mark.parametrize("value", [float("nan"), float("inf")]) +def test_non_finite_floats_are_rejected(value: float) -> None: + with pytest.raises(ValueError): + to_wire_value(value) + + +def test_unsupported_types_and_non_object_content_are_rejected() -> None: + with pytest.raises(TypeError): + to_wire_value(object()) + with pytest.raises(TypeError): + to_wire_value({1: "a"}) + with pytest.raises(TypeError): + to_wire_value(b"bytes") + with pytest.raises(TypeError, match="mapping or a dataclass"): + serialize_content([1, 2])