diff --git a/CHANGELOG.md b/CHANGELOG.md index d3311227..3784f125 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,81 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). --- +## v26.09.07 (2026-09-24) + +Three asks a platform raised against `26.09.06` on the day it shipped: logs that could not be +searched by the identifiers the product is built around, a serving process that had to own the +framework's own tables to start, and a cache bridge that made every process a consumer of a +queue it had nothing to do with. Each was found by running a service on the framework, not by +reading it. + +### Fixed + +- **An identifier in a log message survives redaction, and `PHONE` means a phone number.** The + built-in `PHONE` rule was guarded by digits only — `(?b2aa0`. A run id grepped out of an application therefore never + matched its own log line — silently, and only sometimes, which is worse than always, because + the operator concludes the event never happened. The same rule could not match compact E.164 + at all (eleven contiguous digits cannot be split by `\d{3}\d{4}`), so `+34911234567` went to + the log untouched: it gave up searchability *and* privacy at once, and neither is traded away + to fix the other. Identifier shapes — uuid of any version, W3C trace id and span id, ULID — + are now carved out of the message **before** any entity pattern runs, and the entity rules + only ever see the gaps between them, so no rule can consume half of one, on the regex engine + and on the Presidio engine alike. This is the message-text half of the rule `#156` applied to + the structlog correlation *field keys*; the 16- and 32-character hex shapes require at least + one hex letter on purpose, so an all-digit run of that length is still read as a card number + rather than hidden behind the guard. `PHONE` itself is now anchored on word boundaries and + matches compact E.164, separated international, `(212) 555-0143`, `202-555-0143` and + `555-0143`; a national number written as seven bare digits with no separator is deliberately + no longer matched. `pyfly.logging.redaction.preserve-patterns` (a named-regex map, symmetric + with `extra-patterns`) protects a consumer's own id shape, and an invalid entry there is + warned about and ignored exactly as an invalid `extra-patterns` entry is. Forty tests in + `tests/logging/test_identifiers_survive_redaction.py`, including a sweep over three thousand + generated uuids per version — with a quarter of them mangled before, a handful of examples + would have passed by luck. + +- **A serving process boots the Postgres event bus with no right to create schema.** + `PostgresEventBus.start()` ran the outbox `CREATE TABLE/INDEX IF NOT EXISTS` in every process + at every boot, including a pure publisher, since `publish()` lazily starts the bus. PostgreSQL + checks `CREATE` on the schema *before* it checks `IF NOT EXISTS`, so an existing table did not + save a role without that right; and `CREATE INDEX IF NOT EXISTS` on a table you do not own + fails even when the index is already there, which a `GRANT CREATE ON SCHEMA` does not fix. The + framework was not asking for one privilege, it was asking every deployment to keep a boot role + that *owned* two framework-internal tables and to make the serving role a member of it — + under a least-privilege posture, not an inconvenience but a violation. `start()` now probes + first with a single `to_regclass` round trip, which needs nothing beyond `USAGE` on the schema, + and issues the DDL only when a table is genuinely missing; a first boot against an empty + database still creates everything, so `pyfly new` needs no configuration. The gate is an + opt-*out*: `pyfly.eda.postgres.auto-create-tables: false` stops the framework issuing DDL under + any circumstance, for a deployment whose schema belongs to its migrations. The consumer group's + cursor row is deliberately outside the skip — it is data, not schema, it needs only `INSERT`, + and a new group joining an existing deployment still has to create its own. Six unit tests in + `tests/eda/test_postgres_outbox_ddl_gate.py` drive `start()` against a fake asyncpg and assert + on the statements it sends; three in `tests/integration/test_eda_postgres_least_privilege.py` + boot the bus against a real server as a role that holds `USAGE`, `SELECT/INSERT/UPDATE` and + nothing else, and fail on the old adapter with `permission denied for schema public`. + +- **The CQRS cache-invalidation bridge subscribes one handler per rule, not the wildcard.** + `EdaCacheInvalidationBridge.subscribe()` registered `"*"` the moment a CQRS context had an EDA + bus, whether or not any rule was registered — so a process subscribed to everything in order + to discard it, since an unregistered event type reaches `on_envelope` and does nothing at all. + On the Postgres bus that is not passive: `_drain` refuses to advance the consumer group's + cursor while no handler is registered, precisely so events published before a worker subscribed + are not lost, and a wildcard handler that dispatches nothing defeated that guard — an API + sharing a worker's consumer group drained the worker's queue into a no-op and its jobs stayed + queued. On Kafka and Redis the same subscription cost a partition assignment and a pattern + subscription for no work. The bridge now subscribes an event type when a rule for it is + registered, and nothing while it has no rule, so a process that invalidates nothing is not a + consumer of the bus and the `_drain` guard works as written; registering before or after the + framework attaches the bridge both work. There was also no way to refuse it short of disabling + CQRS entirely — `pyfly.cqrs.cache.invalidation.enabled: false` now does exactly that. Eleven + tests in `tests/cqrs/test_cache_bridge_subscribes_narrowly.py` and the regression that names + the damage in `tests/eda/test_cursor_guard_with_cache_bridge.py`. + +--- + ## v26.09.06 (2026-09-24) > Prepared during the wave of 2026-09-17 as `26.09.05`; that number was taken meanwhile by the diff --git a/README.md b/README.md index b4b3541c..7062ee04 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ Firefly Framework Python 3.12+ License: Apache 2.0 - Version: 26.09.06 + Version: 26.09.07 Type Checked: mypy strict Code Style: Ruff Async First @@ -852,13 +852,13 @@ See **[`samples/lumen/`](samples/lumen/README.md)** for an end-to-end DDD micros ```bash # Install the latest release (uv) -uv add "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.6-py3-none-any.whl" +uv add "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.7-py3-none-any.whl" # Install with specific extras -uv add "pyfly[web,data-relational,cache] @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.6-py3-none-any.whl" +uv add "pyfly[web,data-relational,cache] @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.7-py3-none-any.whl" # Or with pip -pip install "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.6-py3-none-any.whl" +pip install "pyfly @ https://github.com/fireflyframework/fireflyframework-pyfly/releases/latest/download/pyfly-26.9.7-py3-none-any.whl" ``` ### One-Line Install (CLI + Framework) diff --git a/docs/cli.md b/docs/cli.md index bf80ad84..cc856507 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -775,7 +775,7 @@ Missing optional tools are shown with a `-` dash indicator (dimmed), while missi Verifies that PyFly itself is importable and displays the installed version: ``` -✓ pyfly v26.09.06 +✓ pyfly v26.09.07 ``` ### Summary diff --git a/docs/getting-started.md b/docs/getting-started.md index 55659983..c260e886 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -480,7 +480,7 @@ ______ ___.__._/ ____\ | ___.__. | __// ____| |__| |____/ ____| |__| \/ \/ - PyFly v26.09.06 | Python 3.12.0 + PyFly v26.09.07 | Python 3.12.0 2026-01-15T10:30:00Z [info] starting_application app=my-service version=0.1.0 2026-01-15T10:30:00Z [info] no_active_profiles message=No active profiles set, falling back to default diff --git a/docs/installation.md b/docs/installation.md index c1ac93e4..ff6fb98b 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -512,7 +512,7 @@ PyFly Doctor ✓ mypy — Type checker PyFly packages: - ✓ pyfly v26.09.06 + ✓ pyfly v26.09.07 All checks passed! ``` diff --git a/docs/modules/core.md b/docs/modules/core.md index 920d01c4..8611c5c1 100644 --- a/docs/modules/core.md +++ b/docs/modules/core.md @@ -598,7 +598,7 @@ class BannerMode(enum.Enum): | Mode | Behavior | |---|---| | `TEXT` | Full ASCII art banner (default) with a framework version line. | -| `MINIMAL` | Single line: `:: PyFly :: (v26.09.06)` | +| `MINIMAL` | Single line: `:: PyFly :: (v26.09.07)` | | `OFF` | No banner output at all. | ### BannerPrinter Class @@ -639,7 +639,7 @@ ______ ___.__._/ ____\ | ___.__. | __// ____| |__| |____/ ____| |__| \/ \/ -:: PyFly Framework :: (v26.09.06) +:: PyFly Framework :: (v26.09.07) ``` ### Custom Banner Files diff --git a/docs/modules/cqrs.md b/docs/modules/cqrs.md index 715f818d..d24aac30 100644 --- a/docs/modules/cqrs.md +++ b/docs/modules/cqrs.md @@ -509,9 +509,28 @@ and injected into the `DefaultQueryBus`. No extra configuration is required. ### EDA-driven cache invalidation When an EDA `EventPublisher` bean is present, `CqrsAutoConfiguration` -also creates an **`EdaCacheInvalidationBridge`** bean and subscribes it to the -bus as a wildcard listener. The bridge evicts `QueryCacheAdapter` entries in -response to domain events arriving on the `pyfly.eda` bus. +also creates an **`EdaCacheInvalidationBridge`** bean and attaches it to the +bus. The bridge evicts `QueryCacheAdapter` entries in response to domain events +arriving on the `pyfly.eda` bus. + +It subscribes **one handler per registered rule**, and nothing at all while no +rule is registered — a process that invalidates nothing is not a consumer of the +bus. Registering a rule after startup subscribes it there and then, so the order +of `register()` and the framework's `subscribe()` does not matter. + +> **Prior behaviour (corrected in 26.09.07):** the bridge subscribed the +> wildcard `"*"` from every process the moment a CQRS context had an EDA bus, +> whether or not any rule was registered — so a process subscribed to everything +> in order to discard it. On the Postgres bus that is not passive: the adapter +> refuses to advance the consumer group's cursor while no handler is registered, +> precisely so events published before a worker subscribes are not lost, and a +> wildcard handler that dispatches nothing defeated that guard. An API sharing a +> worker's consumer group drained the worker's queue into a no-op and its jobs +> stayed queued. On Kafka and Redis the same subscription cost a partition +> assignment and a pattern subscription for no work. + +Set `pyfly.cqrs.cache.invalidation.enabled: false` to refuse the bridge outright +without disabling CQRS. Register invalidation rules on the bridge after startup (or inject the bean): @@ -684,6 +703,7 @@ pyfly: | `pyfly.cqrs.query.cache_ttl` | `int` | `900` | Default cache TTL (seconds). | | `pyfly.cqrs.query.metrics_enabled` | `bool` | `true` | Query metrics. | | `pyfly.cqrs.query.tracing_enabled` | `bool` | `true` | Query tracing. | +| `pyfly.cqrs.cache.invalidation.enabled` | `bool` | `true` | Wire the EDA-driven cache-invalidation bridge. `false` refuses the bean without disabling CQRS. | | `pyfly.cqrs.authorization.enabled` | `bool` | `true` | Authorization checks. | | `pyfly.cqrs.authorization.custom.enabled` | `bool` | `true` | Custom authorization. | | `pyfly.cqrs.authorization.custom.timeout_ms` | `int` | `5000` | Custom auth timeout. | @@ -709,7 +729,7 @@ these beans into the DI container: | `command_event_publisher` | `CommandEventPublisher` | `EdaCommandEventPublisher` when an EDA `EventPublisher` bean is present; `NoOpEventPublisher` otherwise | | `command_bus` | `DefaultCommandBus` | | | `query_cache_adapter` | `QueryCacheAdapter` | Injects `CacheAdapter` when available; no-op otherwise | -| `eda_cache_invalidation_bridge` | `EdaCacheInvalidationBridge \| None` | Created and subscribed to the EDA bus when an `EventPublisher` bean is present; `None` otherwise | +| `eda_cache_invalidation_bridge` | `EdaCacheInvalidationBridge \| None` | Created and attached to the EDA bus when an `EventPublisher` bean is present, subscribing one event type per registered rule; `None` otherwise. Gated by `pyfly.cqrs.cache.invalidation.enabled` (default `true`) | | `query_bus` | `DefaultQueryBus` | | `cqrs_metrics_service` optionally injects a `MetricsRegistry` bean from the diff --git a/docs/modules/events.md b/docs/modules/events.md index 7d7a7a16..e3307411 100644 --- a/docs/modules/events.md +++ b/docs/modules/events.md @@ -254,6 +254,7 @@ property. All keys are optional; the defaults work for local development. | `pyfly.eda.postgres.dsn` | `str` | *(required)* | PostgreSQL DSN for the producer connection pool. | | `pyfly.eda.postgres.listen-dsn` | `str` | same as `dsn` | Optional dedicated DSN for the LISTEN connection. | | `pyfly.eda.postgres.channel` | `str` | `pyfly_eda` | `pg_notify` channel name. | +| `pyfly.eda.postgres.auto-create-tables` | `bool` | `true` | Create `pyfly_eda_outbox` / `pyfly_eda_offsets` when they are missing. The DDL is skipped when they already exist, so a serving process needs no schema-creation right; `false` means the framework never issues DDL at all. | | `pyfly.eda.rabbitmq.url` | `str` | `amqp://guest:guest@localhost/` | AMQP connection URL. | | `pyfly.eda.rabbitmq.exchange-name` | `str` | `pyfly` | Name of the durable DIRECT exchange to declare. | @@ -295,6 +296,34 @@ The JSON serializer **reads both envelope shapes** of the Firefly family — PyF empty object) — and raises one typed `EnvelopeDecodeError` for anything else. To *write* the LaraFly shape (its key order, `DATE_ATOM` timestamps), select `serialization-format: firefly-json`. +### Postgres: what privileges a serving process actually needs + +`PostgresEventBus.start()` used to run `CREATE TABLE IF NOT EXISTS` and +`CREATE INDEX IF NOT EXISTS` for the outbox in **every** process at **every** boot — including a +process that only publishes, since `publish()` lazily starts the bus. Postgres checks `CREATE` on +the schema *before* it checks `IF NOT EXISTS`, so an existing table did not save a role without +that right; and `CREATE INDEX IF NOT EXISTS` on a table you do not own fails even when the index +is already there, which a `GRANT CREATE ON SCHEMA` does not fix. The practical consequence was +that a serving role had to own two framework-internal tables, or belong to a role that did. + +`start()` now **probes first**: one `to_regclass` round trip, which needs nothing beyond `USAGE` +on the schema, and the DDL runs only when a table is genuinely missing. A first boot against an +empty database still creates everything, so `pyfly new` needs no configuration. Set +`pyfly.eda.postgres.auto-create-tables: false` when migrations own the schema and the framework +must not issue DDL under any circumstance. + +A serving process therefore needs only: + +```sql +GRANT USAGE ON SCHEMA public TO serving; +GRANT SELECT, INSERT, UPDATE ON pyfly_eda_outbox, pyfly_eda_offsets TO serving; +GRANT USAGE ON SEQUENCE pyfly_eda_outbox_id_seq TO serving; +``` + +The consumer group's cursor row (`INSERT ... ON CONFLICT DO NOTHING` into `pyfly_eda_offsets`) is +issued on every boot regardless: it is data, not schema, it needs only `INSERT`, and a new +consumer group joining an existing deployment still has to create its own. + --- ## Declarative Decorators diff --git a/docs/modules/logging.md b/docs/modules/logging.md index fb6befbf..faf1debc 100644 --- a/docs/modules/logging.md +++ b/docs/modules/logging.md @@ -194,6 +194,7 @@ message for sensitive entities before writing to any output. | `redaction.entities` | list[string] | see below | Entity names to detect and mask | | `redaction.mask` | string | `placeholder` | Mask style: `placeholder` (``), `partial` (`****1111`), or `hash` (``) | | `redaction.extra-patterns` | map[string, string] | `{}` | Additional named regex patterns to redact (e.g. `{"EMP": "EMP-\\d{4}"}`) | +| `redaction.preserve-patterns` | map[string, string] | `{}` | Additional named regex patterns whose matches are **never** redacted, on top of the built-in identifier shapes (e.g. `{"ORDER_REF": "ORD-[0-9]{8}"}`) | | `redaction.deny-fields` | list[string] | `["password", "token", "secret"]` | Structured log fields whose value is always replaced with `` | | `redaction.allow-fields` | list[string] | `[]` | When set, only these structured fields (plus `event`) are scanned; others are skipped | | `redaction.streams.enabled` | bool | `false` | Opt-in: wrap `sys.stdout`/`sys.stderr` with the redactor | @@ -208,6 +209,44 @@ Default entities detected by the regex engine: ship but are **off by default** — IP addresses are common in logs and redacting them is often undesirable; enable them via `redaction.entities` if you need them.) +### Identifiers are never mangled + +A uuid, a W3C trace id or span id, and a ULID are made of digits and hex letters, +so a generic PII rule can match a substring of one. Before 26.09.07 that is +exactly what happened: roughly one uuid in four had a digit run in its middle +replaced by ``, so a run id grepped out of an application never matched +the log line it appeared on — silently, and only sometimes. + +Those shapes are now carved out of the message **before** any entity pattern +runs, and the entity rules only ever see the gaps between them. A pattern +therefore cannot match across a protected identifier, whichever engine is in +use. The built-in protected shapes are: + +| Name | Shape | +|------|-------| +| `UUID` | `xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`, any version, either case | +| `TRACE_ID` | 32 hex characters (W3C trace-context), at least one of them a letter | +| `SPAN_ID` | 16 hex characters (W3C trace-context), at least one of them a letter | +| `ULID` | 26 Crockford base32 characters starting `0`-`7` | + +The hex shapes require at least one hex **letter** on purpose: an all-digit run +of 16 characters is a card number far more often than it is a span id, and +protecting it would turn this guard into a way to hide PII. Add your own +identifier shapes with `redaction.preserve-patterns`. + +This is the message-text half of the rule +`pyfly.logging.redaction.processor._NEVER_REDACT` applies to the structlog +correlation *fields* (`trace_id`, `span_id`, `trace_flags`, `correlationId`). + +`PHONE` was tightened in the same release. It used to be guarded by digits only +(`(? None: self._cache = cache self._rules: dict[str, list[str]] = {} + self._publisher: EventPublisher | None = None + self._subscribed: set[str] = set() # ── rule registration ────────────────────────────────────── def register(self, event_type: str, cache_key_pattern: str) -> None: """Register a cache-key pattern to evict when *event_type* arrives. + When the bridge is already attached to a bus, registering a rule also + subscribes that one event type — the bridge only ever consumes what it + has something to do about. + Args: event_type: The EDA event-type string (e.g. ``"order.updated"``). cache_key_pattern: A pattern such as ``"order:{order_id}"`` where ``{field}`` placeholders are resolved from the envelope payload. """ self._rules.setdefault(event_type, []).append(cache_key_pattern) + self._subscribe_event_type(event_type) # ── EDA subscription ─────────────────────────────────────── def subscribe(self, event_publisher: EventPublisher) -> None: - """Wire this bridge into the EDA bus. + """Attach this bridge to the EDA bus. + + Subscribes :meth:`on_envelope` **once per registered event type**, and + nothing at all while no rule is registered. + + It used to subscribe the wildcard ``"*"``, which made every process a + consuming member of the group in order to discard what it received. On + the Postgres bus that is not passive: the adapter refuses to advance its + cursor while no handler is registered, precisely so events published + before a worker subscribed are not lost — and a wildcard handler that + dispatches nothing defeated that guard, draining another process's queue + into a no-op. On Kafka and Redis it cost a partition assignment and a + pattern subscription for no work. - Subscribes :meth:`on_envelope` as a wildcard handler so it receives - **every** event. The bridge then routes internally based on the - registered rules. + Attaching before or after :meth:`register` both work: rules registered + later subscribe themselves. Args: event_publisher: The live EDA bus (e.g. an :class:`~pyfly.eda.adapters.memory.InMemoryEventBus`). """ - event_publisher.subscribe("*", self.on_envelope) + self._publisher = event_publisher + for event_type in self._rules: + self._subscribe_event_type(event_type) + + def _subscribe_event_type(self, event_type: str) -> None: + if self._publisher is None or event_type in self._subscribed: + return + self._publisher.subscribe(event_type, self.on_envelope) + self._subscribed.add(event_type) # ── event handler ────────────────────────────────────────── diff --git a/src/pyfly/cqrs/config/auto_configuration.py b/src/pyfly/cqrs/config/auto_configuration.py index ee00ab73..9d80fa93 100644 --- a/src/pyfly/cqrs/config/auto_configuration.py +++ b/src/pyfly/cqrs/config/auto_configuration.py @@ -143,6 +143,7 @@ def query_cache_adapter(self, cache: CacheAdapter | None = None) -> QueryCacheAd return QueryCacheAdapter(cache=cache) @bean + @conditional_on_property("pyfly.cqrs.cache.invalidation.enabled", having_value="true", match_if_missing=True) def eda_cache_invalidation_bridge( self, cache: QueryCacheAdapter, @@ -151,19 +152,22 @@ def eda_cache_invalidation_bridge( """Wire the EDA→CQRS cache-invalidation bridge. When an :class:`~pyfly.eda.ports.outbound.EventPublisher` bean is - present the bridge is created, subscribed to the bus and returned so - that applications can register additional rules via - ``bridge.register(event_type, pattern)``. + present the bridge is created, attached to the bus and returned so + that applications can register rules via + ``bridge.register(event_type, pattern)``. Attaching subscribes one + handler per registered rule and nothing while there is no rule, so a + process that invalidates nothing is not a consumer of the bus. When no EDA bus is configured the bean evaluates to ``None`` and the - bridge is silently disabled. + bridge is silently disabled. ``pyfly.cqrs.cache.invalidation.enabled: + false`` refuses it outright, without disabling CQRS. """ if producer is None: _logger.debug("No EventPublisher bean found — EDA cache-invalidation bridge disabled") return None bridge = EdaCacheInvalidationBridge(cache) bridge.subscribe(producer) - _logger.debug("EDA cache-invalidation bridge wired to %s", type(producer).__name__) + _logger.debug("EDA cache-invalidation bridge attached to %s", type(producer).__name__) return bridge @bean diff --git a/src/pyfly/cqrs/config/properties.py b/src/pyfly/cqrs/config/properties.py index e9252ae0..cb47bff6 100644 --- a/src/pyfly/cqrs/config/properties.py +++ b/src/pyfly/cqrs/config/properties.py @@ -63,6 +63,20 @@ class QueryProperties: tracing_enabled: bool = True +@dataclass +class CacheInvalidationProperties: + """``pyfly.cqrs.cache.invalidation.*``.""" + + enabled: bool = True + + +@dataclass +class CqrsCacheProperties: + """``pyfly.cqrs.cache.*``.""" + + invalidation: CacheInvalidationProperties = field(default_factory=CacheInvalidationProperties) + + @dataclass class CustomAuthorizationProperties: """``pyfly.cqrs.authorization.custom.*``.""" @@ -87,4 +101,5 @@ class CqrsProperties: enabled: bool = True command: CommandProperties = field(default_factory=CommandProperties) query: QueryProperties = field(default_factory=QueryProperties) + cache: CqrsCacheProperties = field(default_factory=CqrsCacheProperties) authorization: AuthorizationProperties = field(default_factory=AuthorizationProperties) diff --git a/src/pyfly/eda/adapters/postgres.py b/src/pyfly/eda/adapters/postgres.py index 35205405..b69bf343 100644 --- a/src/pyfly/eda/adapters/postgres.py +++ b/src/pyfly/eda/adapters/postgres.py @@ -124,6 +124,20 @@ def _normalise_dsn(dsn: str) -> str: ); """ +# Postgres checks CREATE on the schema BEFORE it checks IF NOT EXISTS, so +# replaying the DDL above is not free for a role that merely reads and writes +# the two tables: it fails with "permission denied for schema public" even +# though both tables are already there. CREATE INDEX IF NOT EXISTS is stricter +# still — it wants ownership of the table, so a GRANT CREATE ON SCHEMA does not +# rescue it either. That is why a serving process used to need to be (or to +# belong to) the owner of two framework-internal tables. This probe costs one +# round trip, needs nothing beyond USAGE on the schema, and answers the only +# question the DDL was asking. +_SQL_TABLES_PRESENT = """ +SELECT to_regclass('pyfly_eda_outbox') IS NOT NULL + AND to_regclass('pyfly_eda_offsets') IS NOT NULL +""" + class PostgresEventBus: """``EventPublisher`` backed by Postgres LISTEN/NOTIFY + outbox table.""" @@ -137,6 +151,7 @@ def __init__( destinations: list[str] | None = None, group: str = "default", poll_interval_s: float = 5.0, + auto_create_tables: bool = True, ) -> None: # asyncpg only understands the bare ``postgresql://`` scheme; strip # SQLAlchemy-style dialect markers (``+asyncpg``, ``+psycopg``) @@ -147,6 +162,7 @@ def __init__( self._destinations = list(destinations) if destinations else None self._group = group self._poll_interval_s = poll_interval_s + self._auto_create_tables = auto_create_tables self._handlers: list[tuple[str, EventHandler]] = [] self._pool: Any = None self._listen_conn: Any = None @@ -196,8 +212,10 @@ async def start(self) -> None: self._pool = await asyncpg.create_pool(self._dsn, min_size=1, max_size=10) async with self._pool.acquire() as conn: - await conn.execute(_DDL_OUTBOX) - await conn.execute(_DDL_OFFSETS) + await self._ensure_tables(conn) + # Deliberately outside the skip: this is the consumer group's cursor + # row, not schema. It needs INSERT and nothing more, and a new group + # joining an existing deployment still has to create its own. await conn.execute( """ INSERT INTO pyfly_eda_offsets (consumer_group, last_event_id) @@ -225,6 +243,28 @@ async def start(self) -> None: self._group, ) + async def _ensure_tables(self, conn: Any) -> None: + """Create the outbox tables, but only when they are actually missing. + + Every process used to replay the CREATE TABLE/INDEX statements at every + boot, so a serving role needed schema-creation rights (in practice, + ownership of framework-internal tables) for work it never did. The probe + is one round trip on a connection the adapter has already opened, and it + runs as any role that can read the schema. + """ + if not self._auto_create_tables: + logger.debug( + "pyfly.eda.postgres.auto-create-tables is off; assuming the outbox tables are managed elsewhere" + ) + return + if await conn.fetchval(_SQL_TABLES_PRESENT): + # Say so out loud: an operator chasing a missing table wants to see + # that the framework looked before it decided to do nothing. + logger.info("EDA outbox tables already present; skipping DDL") + return + await conn.execute(_DDL_OUTBOX) + await conn.execute(_DDL_OFFSETS) + async def stop(self) -> None: self._closed = True self._started = False diff --git a/src/pyfly/eda/auto_configuration.py b/src/pyfly/eda/auto_configuration.py index 98d75f79..5f12f7cc 100644 --- a/src/pyfly/eda/auto_configuration.py +++ b/src/pyfly/eda/auto_configuration.py @@ -41,6 +41,10 @@ * ``postgres.listen-dsn`` — Optional dedicated DSN for the LISTEN connection. Defaults to ``postgres.dsn``. * ``postgres.channel`` — pg_notify channel. Default ``pyfly_eda``. +* ``postgres.auto-create-tables`` — Create the outbox tables when they are + missing. Default ``true``. The DDL is skipped when they already exist, so a + serving process needs no schema-creation right; set this to ``false`` when the + schema is managed by migrations and the framework must never issue DDL at all. * ``rabbitmq.url`` — AMQP URL. Default ``amqp://guest:guest@localhost/``. * ``rabbitmq.exchange-name`` — Exchange name. Default ``pyfly``. """ @@ -132,12 +136,18 @@ def event_publisher(self, config: Config) -> EventPublisher: listen_dsn_raw = config.get("pyfly.eda.postgres.listen-dsn", "") listen_dsn = str(listen_dsn_raw) if listen_dsn_raw else None channel = str(config.get("pyfly.eda.postgres.channel", "pyfly_eda")) + auto_create = str(config.get("pyfly.eda.postgres.auto-create-tables", "true")).lower() in ( + "true", + "1", + "yes", + ) return PostgresEventBus( dsn=dsn, listen_dsn=listen_dsn, channel=channel, destinations=destinations, group=group, + auto_create_tables=auto_create, ) if provider == "rabbitmq": diff --git a/src/pyfly/logging/redaction/engine.py b/src/pyfly/logging/redaction/engine.py index 6e4614ba..db92fad9 100644 --- a/src/pyfly/logging/redaction/engine.py +++ b/src/pyfly/logging/redaction/engine.py @@ -21,7 +21,7 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable -from pyfly.logging.redaction.patterns import BUILTIN_PATTERNS, VALIDATORS +from pyfly.logging.redaction.patterns import BUILTIN_PATTERNS, PRESERVE_PATTERNS, VALIDATORS if TYPE_CHECKING: from pyfly.config.properties.logging import RedactionProperties @@ -46,6 +46,56 @@ def _mask(value: str, entity: str, style: str) -> str: return f"<{entity}>" +def compile_preserve_patterns(extra: dict[str, str] | None = None) -> list[re.Pattern[str]]: + """The identifier shapes no entity pattern may touch, built-ins first. + + Each pattern is kept separate rather than joined into one alternation, so a + consumer-supplied regex with named groups or backreferences cannot break the + others. An invalid one is warned about and dropped, exactly as an invalid + ``extra-patterns`` entry is. + """ + patterns = list(PRESERVE_PATTERNS.values()) + for name, regex in (extra or {}).items(): + try: + patterns.append(re.compile(regex)) + except re.error: + _logger.warning("Ignoring invalid redaction preserve pattern for %s", name) + return patterns + + +def protected_spans(text: str, patterns: list[re.Pattern[str]]) -> list[tuple[int, int]]: + """Merged, ordered ``(start, end)`` ranges of *text* that must survive redaction.""" + found: list[tuple[int, int]] = [] + for pattern in patterns: + found.extend((m.start(), m.end()) for m in pattern.finditer(text) if m.end() > m.start()) + if not found: + return found + found.sort() + merged: list[tuple[int, int]] = [found[0]] + for start, end in found[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def _redact_around_protected(text: str, patterns: list[re.Pattern[str]], redact_gap: Callable[[str], str]) -> str: + """Apply *redact_gap* to everything in *text* except the protected spans.""" + spans = protected_spans(text, patterns) + if not spans: + return redact_gap(text) + parts: list[str] = [] + cursor = 0 + for start, end in spans: + parts.append(redact_gap(text[cursor:start])) + parts.append(text[start:end]) + cursor = end + parts.append(redact_gap(text[cursor:])) + return "".join(parts) + + class RegexRedactor: """Pattern-based redactor — fast, no external dependencies (default).""" @@ -54,6 +104,7 @@ def __init__( entities: list[str], mask: str = "placeholder", extra_patterns: dict[str, str] | None = None, + preserve_patterns: dict[str, str] | None = None, ) -> None: self._mask = mask self._rules: list[tuple[str, re.Pattern[str], Callable[[str], bool] | None]] = [] @@ -66,10 +117,20 @@ def __init__( self._rules.append((name, re.compile(regex), None)) except re.error: _logger.warning("Ignoring invalid redaction pattern for %s", name) + self._preserve = compile_preserve_patterns(preserve_patterns) def redact(self, text: Any) -> Any: if not isinstance(text, str) or not text: return text + # Identifier shapes are carved out FIRST and the entity rules only ever + # see the gaps between them. Splitting the text is what makes the + # protection absolute: a pattern cannot match across a protected span, + # so no rule can consume half a uuid. + return _redact_around_protected(text, self._preserve, self._redact_entities) + + def _redact_entities(self, text: str) -> str: + if not text: + return text result = text for entity, pattern, validator in self._rules: @@ -114,23 +175,34 @@ def __init__(self, props: RedactionProperties) -> None: self._threshold = props.presidio.score_threshold # Regex pass runs after presidio to also mask token-types presidio has no # recognizer for (JWT, bearer tokens, URL credentials). - self._fallback = RegexRedactor(props.entities, props.mask, props.extra_patterns) + self._fallback = RegexRedactor(props.entities, props.mask, props.extra_patterns, props.preserve_patterns) + self._preserve = compile_preserve_patterns(props.preserve_patterns) def redact(self, text: Any) -> Any: if not isinstance(text, str) or not text: return text - result = text + # Presidio's recognizers are as capable of reading a uuid as a phone + # number as the regex engine was, so it too only ever sees the gaps + # between the protected identifier spans. + result = _redact_around_protected(text, self._preserve, self._analyze_and_anonymize) + # Always run the regex pass too, for the token-types presidio misses. + # It protects the same spans again on its own. + return self._fallback.redact(result) + + def _analyze_and_anonymize(self, text: str) -> str: + if not text: + return text try: # Detect with presidio's full recognizer set (its entity names differ # from the regex engine's, so we do NOT restrict to props.entities). - findings = self._analyzer.analyze(text=result, language=self._language) + findings = self._analyzer.analyze(text=text, language=self._language) findings = [r for r in findings if r.score >= self._threshold] if findings: - result = self._anonymizer.anonymize(text=result, analyzer_results=findings).text + anonymized: str = self._anonymizer.anonymize(text=text, analyzer_results=findings).text + return anonymized except Exception: # noqa: BLE001 — never let redaction crash logging pass - # Always run the regex pass too, for the token-types presidio misses. - return self._fallback.redact(result) + return text def build_redactor(props: RedactionProperties) -> Redactor | None: @@ -150,4 +222,4 @@ def build_redactor(props: RedactionProperties) -> Redactor | None: _logger.warning("presidio PII engine unavailable (%s); falling back to regex redaction", exc) else: _logger.info("presidio not available (%s); using regex PII redaction (install pyfly[pii])", exc) - return RegexRedactor(props.entities, props.mask, props.extra_patterns) + return RegexRedactor(props.entities, props.mask, props.extra_patterns, props.preserve_patterns) diff --git a/src/pyfly/logging/redaction/patterns.py b/src/pyfly/logging/redaction/patterns.py index 301ea52f..843a9a78 100644 --- a/src/pyfly/logging/redaction/patterns.py +++ b/src/pyfly/logging/redaction/patterns.py @@ -46,7 +46,21 @@ def luhn_valid(value: str) -> bool: "JWT": re.compile(r"\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b"), "BEARER_TOKEN": re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._\-]+"), "URL_CREDENTIALS": re.compile(r"://[^/\s:@]+:([^/\s:@]+)@"), - "PHONE": re.compile(r"(? bool: VALIDATORS: dict[str, Callable[[str], bool]] = { "CREDIT_CARD": luhn_valid, } + + +# Identifier shapes that are NEVER PII and are therefore protected from every +# entity pattern above. A uuid, a W3C trace id and a ULID are digits and hex +# letters by construction, so a generic PII rule can match a substring of one +# and mangle it — silently, and only sometimes, which is worse than mangling it +# always: a run id grepped out of an application never matches the log line, and +# the operator concludes the event never happened. ``processor._NEVER_REDACT`` +# already applies this principle to the structlog correlation FIELDS; these +# patterns apply it to the identifiers carried in the message TEXT, wherever +# they appear. +# +# The 16- and 32-char hex shapes require at least one hex LETTER on purpose: an +# all-digit run of that length is a credit-card number far more often than it is +# a span id, and protecting it would turn this guard into a way to hide PII. +_HEX_ID = r"\b(?=[0-9a-fA-F]{{{n}}}\b)[0-9a-fA-F]*[a-fA-F][0-9a-fA-F]*\b" + +PRESERVE_PATTERNS: dict[str, re.Pattern[str]] = { + # uuid, any version, either case + "UUID": re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b"), + # W3C trace-context trace id (32 hex) and span id (16 hex) + "TRACE_ID": re.compile(_HEX_ID.format(n=32)), + "SPAN_ID": re.compile(_HEX_ID.format(n=16)), + # ULID — Crockford base32, 26 chars, first char 0-7 + "ULID": re.compile(r"\b[0-7][0-9A-HJKMNP-TV-Z]{25}\b"), +} diff --git a/tests/cqrs/test_cache_bridge_subscribes_narrowly.py b/tests/cqrs/test_cache_bridge_subscribes_narrowly.py new file mode 100644 index 00000000..6ed0dc8d --- /dev/null +++ b/tests/cqrs/test_cache_bridge_subscribes_narrowly.py @@ -0,0 +1,157 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The cache-invalidation bridge consumes only what it has a rule for. + +It used to subscribe ``"*"`` from every process the moment a CQRS context had an +EDA bus, whether or not any rule was registered — so a process subscribed to +everything in order to discard it, and on a durable bus that made it a consuming +member of the group. +""" + +from __future__ import annotations + +import pytest + +from pyfly.cache.adapters.memory import InMemoryCache +from pyfly.container.container import Container +from pyfly.context.condition_evaluator import ConditionEvaluator +from pyfly.core.config import Config +from pyfly.cqrs.cache.adapter import QueryCacheAdapter +from pyfly.cqrs.cache.eda_bridge import EdaCacheInvalidationBridge +from pyfly.cqrs.config.auto_configuration import CqrsAutoConfiguration +from pyfly.cqrs.config.properties import CqrsProperties +from pyfly.eda.adapters.memory import InMemoryEventBus +from pyfly.eda.types import EventEnvelope + + +class RecordingPublisher: + """An ``EventPublisher`` that only remembers what was subscribed to it.""" + + def __init__(self) -> None: + self.subscriptions: list[str] = [] + + def subscribe(self, event_type_pattern: str, handler: object) -> None: + self.subscriptions.append(event_type_pattern) + + async def publish( + self, + destination: str, + event_type: str, + payload: dict[str, object], + headers: dict[str, str] | None = None, + ) -> None: + return None + + +def bridge() -> EdaCacheInvalidationBridge: + return EdaCacheInvalidationBridge(QueryCacheAdapter(cache=None)) + + +class TestSubscriptions: + def test_a_bridge_with_no_rule_subscribes_to_nothing(self) -> None: + publisher = RecordingPublisher() + bridge().subscribe(publisher) + assert publisher.subscriptions == [] + + def test_a_registered_rule_subscribes_that_event_type_and_not_the_wildcard(self) -> None: + publisher = RecordingPublisher() + b = bridge() + b.register("order.updated", "order:{order_id}") + b.subscribe(publisher) + assert publisher.subscriptions == ["order.updated"] + + def test_a_rule_registered_after_attaching_subscribes_itself(self) -> None: + publisher = RecordingPublisher() + b = bridge() + b.subscribe(publisher) + assert publisher.subscriptions == [] + b.register("order.updated", "order:{order_id}") + assert publisher.subscriptions == ["order.updated"] + + def test_a_second_event_type_adds_a_second_subscription(self) -> None: + publisher = RecordingPublisher() + b = bridge() + b.register("order.updated", "order:{order_id}") + b.register("order.deleted", "order:{order_id}") + b.subscribe(publisher) + assert publisher.subscriptions == ["order.updated", "order.deleted"] + + def test_two_patterns_for_one_event_type_subscribe_once(self) -> None: + publisher = RecordingPublisher() + b = bridge() + b.register("order.updated", "order:{order_id}") + b.register("order.updated", "orders:list") + b.subscribe(publisher) + assert publisher.subscriptions == ["order.updated"] + + +@pytest.mark.asyncio +async def test_a_narrow_subscription_still_evicts_over_a_real_bus() -> None: + """The behaviour the wildcard was there for, over an InMemoryEventBus.""" + cache = InMemoryCache() + await cache.start() + adapter = QueryCacheAdapter(cache=cache) + await adapter.put("order:42", {"id": "42"}, ttl=300) + + bus = InMemoryEventBus() + await bus.start() + b = EdaCacheInvalidationBridge(adapter) + b.register("order.updated", "order:{order_id}") + b.subscribe(bus) + + await bus.publish("pyfly.events", "order.updated", {"order_id": "42"}) + assert await adapter.get("order:42") is None + + +@pytest.mark.asyncio +async def test_an_unregistered_event_type_never_reaches_the_bridge() -> None: + """Previously it arrived and was discarded; now it is not delivered at all.""" + seen: list[EventEnvelope] = [] + b = bridge() + + async def spy(envelope: EventEnvelope) -> None: + seen.append(envelope) + + b.on_envelope = spy # type: ignore[method-assign] + bus = InMemoryEventBus() + await bus.start() + b.register("order.updated", "order:{order_id}") + b.subscribe(bus) + + await bus.publish("pyfly.events", "ingest.source.requested", {"source_id": "1"}) + assert seen == [] + + +class TestOffSwitch: + """``pyfly.cqrs.cache.invalidation.enabled`` — refusing the bridge without disabling CQRS.""" + + @staticmethod + def _included(enabled: str | None) -> bool: + raw: dict[str, object] = ( + {} if enabled is None else {"pyfly": {"cqrs": {"cache": {"invalidation": {"enabled": enabled}}}}} + ) + evaluator = ConditionEvaluator(Config(raw), Container()) + return evaluator.should_include_method(CqrsAutoConfiguration.eda_cache_invalidation_bridge) + + def test_the_bean_is_wired_by_default(self) -> None: + assert self._included(None) is True + + def test_the_property_refuses_the_bean(self) -> None: + assert self._included("false") is False + + def test_the_property_set_to_true_keeps_the_bean(self) -> None: + assert self._included("true") is True + + def test_the_property_defaults_to_on(self) -> None: + assert CqrsProperties().cache.invalidation.enabled is True diff --git a/tests/eda/test_cursor_guard_with_cache_bridge.py b/tests/eda/test_cursor_guard_with_cache_bridge.py new file mode 100644 index 00000000..ac69b28e --- /dev/null +++ b/tests/eda/test_cursor_guard_with_cache_bridge.py @@ -0,0 +1,59 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The damage the wildcard subscription did, named. + +``PostgresEventBus._drain`` refuses to advance the group's cursor while no +handler is registered, so events published before a worker subscribed are not +lost. The CQRS cache bridge used to register a wildcard handler in every process +whether or not it had a rule, which made ``_handlers`` non-empty and defeated +that guard: an API process drained the worker's queue into a no-op and the jobs +stayed queued forever. +""" + +from __future__ import annotations + +import pytest + +from pyfly.cqrs.cache.adapter import QueryCacheAdapter +from pyfly.cqrs.cache.eda_bridge import EdaCacheInvalidationBridge +from pyfly.eda.adapters.postgres import PostgresEventBus + + +def _bus_and_bridge() -> tuple[PostgresEventBus, EdaCacheInvalidationBridge]: + bus = PostgresEventBus(dsn="postgresql://x/y", group="workers") + return bus, EdaCacheInvalidationBridge(QueryCacheAdapter(cache=None)) + + +@pytest.mark.asyncio +async def test_a_rule_less_bridge_leaves_the_bus_with_no_handlers() -> None: + bus, bridge = _bus_and_bridge() + bridge.subscribe(bus) + assert bus._handlers == [] + + # _drain() returns on the handler guard before it ever acquires a + # connection — with no pool, reaching the pool would raise. + assert bus._pool is None + await bus._drain() + + +@pytest.mark.asyncio +async def test_a_bridge_with_a_rule_is_a_real_consumer() -> None: + """The contrast: a bridge that has work to do does register, and does drain.""" + bus, bridge = _bus_and_bridge() + bridge.register("order.updated", "order:{order_id}") + bridge.subscribe(bus) + assert [pattern for pattern, _ in bus._handlers] == ["order.updated"] + + with pytest.raises(AttributeError): + await bus._drain() # past the guard, into the (absent) pool diff --git a/tests/eda/test_postgres_outbox_ddl_gate.py b/tests/eda/test_postgres_outbox_ddl_gate.py new file mode 100644 index 00000000..606c2484 --- /dev/null +++ b/tests/eda/test_postgres_outbox_ddl_gate.py @@ -0,0 +1,203 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""A second boot of :class:`PostgresEventBus` issues no DDL. + +The adapter used to replay ``CREATE TABLE/INDEX IF NOT EXISTS`` in every process +at every boot, so a serving role needed schema-creation rights — in practice +ownership of two framework-internal tables — for work it never did. These tests +drive ``start()`` against a fake asyncpg and assert on the statements it sends. +The privilege behaviour itself is proved against a real server in +``tests/integration/test_eda_postgres_least_privilege.py``. +""" + +from __future__ import annotations + +import sys +import types +from typing import Any + +import pytest + +from pyfly.eda.adapters.postgres import PostgresEventBus + + +class FakeConnection: + """Records every statement, and answers the existence probe as told.""" + + def __init__(self, tables_present: bool) -> None: + self.tables_present = tables_present + self.executed: list[str] = [] + self.queried: list[str] = [] + self.listeners: list[str] = [] + + async def execute(self, sql: str, *args: Any) -> None: + self.executed.append(sql.strip()) + + async def fetchval(self, sql: str, *args: Any) -> Any: + self.queried.append(sql.strip()) + if "to_regclass" in sql: + return self.tables_present + if "RETURNING id" in sql: + return 1 + return None + + async def fetch(self, sql: str, *args: Any) -> list[Any]: + return [] + + async def add_listener(self, channel: str, _callback: Any) -> None: + self.listeners.append(channel) + + async def remove_listener(self, channel: str, _callback: Any) -> None: + self.listeners.remove(channel) + + async def close(self) -> None: + return None + + # -- assertions the tests read ------------------------------------ + + @property + def ddl(self) -> list[str]: + return [sql for sql in self.executed if sql.upper().startswith("CREATE")] + + @property + def probes(self) -> list[str]: + return [sql for sql in self.queried if "to_regclass" in sql] + + +class _Acquired: + def __init__(self, conn: FakeConnection) -> None: + self._conn = conn + + async def __aenter__(self) -> FakeConnection: + return self._conn + + async def __aexit__(self, *_exc: Any) -> None: + return None + + +class FakePool: + def __init__(self, conn: FakeConnection) -> None: + self._conn = conn + + def acquire(self) -> _Acquired: + return _Acquired(self._conn) + + async def close(self) -> None: + return None + + +@pytest.fixture +def fake_asyncpg(monkeypatch: pytest.MonkeyPatch) -> Any: + """Install a fake ``asyncpg`` module; the connection is set per test.""" + holder: dict[str, FakeConnection] = {} + + async def create_pool(_dsn: str, **_kwargs: Any) -> FakePool: + return FakePool(holder["conn"]) + + async def connect(_dsn: str, **_kwargs: Any) -> FakeConnection: + return holder["conn"] + + module = types.ModuleType("asyncpg") + module.create_pool = create_pool # type: ignore[attr-defined] + module.connect = connect # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "asyncpg", module) + + def use(conn: FakeConnection) -> FakeConnection: + holder["conn"] = conn + return conn + + return use + + +OFFSETS_INSERT = "INSERT INTO pyfly_eda_offsets" + + +@pytest.mark.asyncio +async def test_a_second_boot_issues_no_ddl(fake_asyncpg: Any) -> None: + conn = fake_asyncpg(FakeConnection(tables_present=True)) + bus = PostgresEventBus(dsn="postgresql://x/y", group="serving") + try: + await bus.start() + finally: + await bus.stop() + + assert conn.ddl == [], f"DDL was issued against existing tables: {conn.ddl}" + assert conn.probes, "the adapter did not probe for the tables" + assert any(sql.startswith(OFFSETS_INSERT) for sql in conn.executed), ( + "the consumer group's cursor row must still be created" + ) + + +@pytest.mark.asyncio +async def test_a_first_boot_still_creates_the_tables(fake_asyncpg: Any) -> None: + conn = fake_asyncpg(FakeConnection(tables_present=False)) + bus = PostgresEventBus(dsn="postgresql://x/y") + try: + await bus.start() + finally: + await bus.stop() + + assert len(conn.ddl) == 2 + assert "CREATE TABLE IF NOT EXISTS pyfly_eda_outbox" in conn.ddl[0] + assert "CREATE INDEX IF NOT EXISTS pyfly_eda_outbox_dest_idx" in conn.ddl[0] + assert "CREATE TABLE IF NOT EXISTS pyfly_eda_offsets" in conn.ddl[1] + assert any(sql.startswith(OFFSETS_INSERT) for sql in conn.executed) + + +@pytest.mark.asyncio +async def test_auto_create_tables_false_neither_probes_nor_creates(fake_asyncpg: Any) -> None: + """Whatever the database holds: the framework issues no DDL at all.""" + conn = fake_asyncpg(FakeConnection(tables_present=False)) + bus = PostgresEventBus(dsn="postgresql://x/y", auto_create_tables=False) + try: + await bus.start() + finally: + await bus.stop() + + assert conn.ddl == [] + assert conn.probes == [] + assert any(sql.startswith(OFFSETS_INSERT) for sql in conn.executed) + + +@pytest.mark.asyncio +async def test_a_publisher_that_lazily_starts_hits_the_same_gate(fake_asyncpg: Any) -> None: + """``publish()`` starts the bus on a bus nobody started — the DDL path too.""" + conn = fake_asyncpg(FakeConnection(tables_present=True)) + bus = PostgresEventBus(dsn="postgresql://x/y") + try: + await bus.publish("pyfly.events", "order.created", {"id": 1}) + finally: + await bus.stop() + + assert conn.ddl == [] + assert any("INSERT INTO pyfly_eda_outbox" in sql for sql in conn.queried) + + +@pytest.mark.asyncio +async def test_the_probe_asks_about_both_tables(fake_asyncpg: Any) -> None: + conn = fake_asyncpg(FakeConnection(tables_present=True)) + bus = PostgresEventBus(dsn="postgresql://x/y") + try: + await bus.start() + finally: + await bus.stop() + + probe = conn.probes[0] + assert "pyfly_eda_outbox" in probe + assert "pyfly_eda_offsets" in probe + + +def test_auto_create_tables_defaults_to_on() -> None: + """A fresh database still just works — the flag is an opt-OUT.""" + assert PostgresEventBus(dsn="postgresql://x/y")._auto_create_tables is True diff --git a/tests/integration/test_eda_postgres_least_privilege.py b/tests/integration/test_eda_postgres_least_privilege.py new file mode 100644 index 00000000..660542cb --- /dev/null +++ b/tests/integration/test_eda_postgres_least_privilege.py @@ -0,0 +1,163 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""A serving process boots the Postgres bus without any right to create schema. + +The finding, written down: `PostgresEventBus.start()` replayed the outbox DDL in +every process at every boot, and Postgres checks `CREATE` on the schema *before* +it checks `IF NOT EXISTS`. A deployment therefore had to keep a boot role that +owned the framework's tables and hand the serving role rights it never used. + +These tests fail at `CREATE TABLE IF NOT EXISTS` on the pre-26.09.07 adapter and +pass on the gated one. Gated by ``@requires_docker``; collected only under +``-m integration``. +""" + +from __future__ import annotations + +import asyncio +import uuid +from collections.abc import AsyncIterator +from urllib.parse import urlsplit, urlunsplit + +import pytest + +from pyfly.eda.adapters.postgres import _DDL_OUTBOX, PostgresEventBus, _normalise_dsn +from pyfly.eda.types import EventEnvelope +from pyfly.testing import requires_docker + +SERVING_PASSWORD = "serving" # noqa: S105 — throwaway role inside a disposable container + + +def _with_credentials(dsn: str, user: str, password: str) -> str: + """Rewrite *dsn*'s userinfo, keeping host, port and database.""" + parts = urlsplit(_normalise_dsn(dsn)) + host = parts.hostname or "localhost" + netloc = f"{user}:{password}@{host}" + if parts.port: + netloc = f"{netloc}:{parts.port}" + return urlunsplit((parts.scheme, netloc, parts.path, "", "")) + + +@pytest.fixture +async def least_privilege_dsn(pg_url: str) -> AsyncIterator[str]: + """Create the outbox tables as their owner, and a role that cannot create schema. + + The role gets exactly what the adapter's real work needs: USAGE on the + schema, SELECT/INSERT/UPDATE on the two tables, USAGE on the sequence. It is + granted no CREATE on the schema and owns nothing. + """ + asyncpg = pytest.importorskip("asyncpg") + role = f"pyfly_serving_{uuid.uuid4().hex[:8]}" + owner_dsn = _normalise_dsn(pg_url) + + conn = await asyncpg.connect(owner_dsn) + try: + # The owner creates the schema, exactly as a migration or a first boot would. + await conn.execute(_DDL_OUTBOX) + await conn.execute( + """ + CREATE TABLE IF NOT EXISTS pyfly_eda_offsets ( + consumer_group TEXT PRIMARY KEY, + last_event_id BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + """ + ) + await conn.execute(f"CREATE ROLE {role} LOGIN PASSWORD '{SERVING_PASSWORD}'") + # Explicit, so the test does not rely on PostgreSQL 15+ having taken + # CREATE on public away from PUBLIC by default. + await conn.execute("REVOKE CREATE ON SCHEMA public FROM PUBLIC") + await conn.execute(f"REVOKE CREATE ON SCHEMA public FROM {role}") + await conn.execute(f"GRANT USAGE ON SCHEMA public TO {role}") + await conn.execute(f"GRANT SELECT, INSERT, UPDATE ON pyfly_eda_outbox, pyfly_eda_offsets TO {role}") + await conn.execute(f"GRANT USAGE ON SEQUENCE pyfly_eda_outbox_id_seq TO {role}") + finally: + await conn.close() + + try: + yield _with_credentials(pg_url, role, SERVING_PASSWORD) + finally: + cleanup = await asyncpg.connect(owner_dsn) + try: + await cleanup.execute(f"REASSIGN OWNED BY {role} TO CURRENT_USER") + await cleanup.execute(f"DROP OWNED BY {role}") + await cleanup.execute(f"DROP ROLE IF EXISTS {role}") + finally: + await cleanup.close() + + +@requires_docker +@pytest.mark.asyncio +async def test_the_ddl_the_adapter_used_to_replay_is_refused(least_privilege_dsn: str) -> None: + """The mechanism, pinned: an existing table does not save a role without CREATE.""" + asyncpg = pytest.importorskip("asyncpg") + conn = await asyncpg.connect(least_privilege_dsn) + try: + with pytest.raises(asyncpg.PostgresError): + await conn.execute(_DDL_OUTBOX) + finally: + await conn.close() + + +@requires_docker +@pytest.mark.asyncio +async def test_a_serving_role_boots_and_round_trips(least_privilege_dsn: str) -> None: + """start() succeeds, and the bus still publishes and consumes.""" + received: list[EventEnvelope] = [] + done = asyncio.Event() + + async def handler(envelope: EventEnvelope) -> None: + received.append(envelope) + done.set() + + group = f"least-privilege-{uuid.uuid4().hex[:8]}" + bus = PostgresEventBus(dsn=least_privilege_dsn, destinations=["pyfly.events"], group=group) + bus.subscribe("order.*", handler) + try: + await bus.start() + await bus.publish("pyfly.events", "order.created", {"id": 1}) + await asyncio.wait_for(done.wait(), timeout=15) + finally: + await bus.stop() + + assert len(received) == 1 + assert received[0].event_type == "order.created" + + +@requires_docker +@pytest.mark.asyncio +async def test_a_first_boot_against_an_empty_database_still_creates_the_tables(pg_url: str) -> None: + """The opt-out default has not cost the zero-configuration first run.""" + asyncpg = pytest.importorskip("asyncpg") + owner_dsn = _normalise_dsn(pg_url) + conn = await asyncpg.connect(owner_dsn) + try: + await conn.execute("DROP TABLE IF EXISTS pyfly_eda_outbox, pyfly_eda_offsets") + finally: + await conn.close() + + bus = PostgresEventBus(dsn=pg_url, group=f"first-boot-{uuid.uuid4().hex[:8]}") + try: + await bus.start() + finally: + await bus.stop() + + conn = await asyncpg.connect(owner_dsn) + try: + present = await conn.fetchval( + "SELECT to_regclass('pyfly_eda_outbox') IS NOT NULL AND to_regclass('pyfly_eda_offsets') IS NOT NULL" + ) + finally: + await conn.close() + assert present is True diff --git a/tests/logging/test_identifiers_survive_redaction.py b/tests/logging/test_identifiers_survive_redaction.py new file mode 100644 index 00000000..a5908462 --- /dev/null +++ b/tests/logging/test_identifiers_survive_redaction.py @@ -0,0 +1,233 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""An identifier in a log message survives redaction; PII in the same line does not. + +The defect these tests pin: the ``PHONE`` rule was guarded by digits only, and a +hex letter or a hyphen satisfies a digit-only guard — so roughly one uuid in four +had a digit run in its middle replaced by ```` and a run id grepped out of +an application never matched its own log line. The same rule could not match +compact E.164, so it gave up searchability AND privacy at once. +""" + +from __future__ import annotations + +import logging +import random +import re +import secrets +import uuid + +import pytest + +from pyfly.config.properties.logging import RedactionProperties +from pyfly.logging.redaction.engine import RegexRedactor, build_redactor, protected_spans +from pyfly.logging.redaction.patterns import BUILTIN_PATTERNS, PRESERVE_PATTERNS +from pyfly.logging.redaction.processor import RedactionFilter + +# A redactor with exactly the framework defaults — what a service gets with no +# configuration at all, which is the configuration the defect was reported under. +DEFAULT_PROPS = RedactionProperties(engine="regex") + + +def default_redactor() -> RegexRedactor: + redactor = build_redactor(DEFAULT_PROPS) + assert isinstance(redactor, RegexRedactor) + return redactor + + +def uuid7() -> str: + """A uuid7-shaped id (48-bit time prefix, version 7, variant 10).""" + raw = bytearray(random.getrandbits(48).to_bytes(6, "big") + secrets.token_bytes(10)) + raw[6] = (raw[6] & 0x0F) | 0x70 + raw[8] = (raw[8] & 0x3F) | 0x80 + return str(uuid.UUID(bytes=bytes(raw))) + + +# (label, token, redacted?) — one row per shape that turns up in a real log line. +SHAPES = [ + ("uuid4", "550e8400-e29b-41d4-a716-446655440000", False), + ("uuid4 upper", "550E8400-E29B-41D4-A716-446655440000", False), + ("uuid7", "01a0b263-b0b7-71f5-bfb2-3b2aa0f1c2d3", False), + ("uuid1", "c232ab00-9414-11ec-b3c8-9f6bdeced846", False), + ("w3c trace id", "4bf92f3577b34da6a3ce929d0e0e4736", False), + ("w3c span id", "00f067aa0ba902b7", False), + ("ulid", "01ARZ3NDEKTSV4RRFFQ69G5FAV", False), + ("iso timestamp", "2026-09-24T07:22:54.123456+00:00", False), + ("sha256", "2d711642b726b04401627ca9fbac32f5c8530fb1903cc4db02258717921a4881", False), + ("base64url token", "HBExZ1mD6Mc6iSGncklO0SvGeXsqj3YTDwjgJRMTmwo", False), + ("ipv4 address", "192.168.1.100", False), + ("e.164 phone (es)", "+34911234567", True), + ("e.164 phone (uk)", "+442071838750", True), + ("e.164 phone (cn)", "+8613800138000", True), + ("international phone", "+1-202-555-0143", True), + ("national phone, spaces", "202 555 0143", True), + ("national phone, dashes", "202-555-0143", True), + ("bracketed area code", "(212) 555-0143", True), + ("16-digit card", "4111111111111111", True), + ("16-digit card, spaced", "4111 1111 1111 1111", True), +] + + +@pytest.mark.parametrize(("label", "token", "redacted"), SHAPES, ids=[row[0] for row in SHAPES]) +def test_shape_is_redacted_only_when_it_is_pii(label: str, token: str, redacted: bool) -> None: + line = f"processing job token={token} ok" + out = default_redactor().redact(line) + if redacted: + assert token not in out, f"{label} leaked through redaction: {out}" + else: + assert out == line, f"{label} was mangled by redaction: {out}" + + +def test_the_reported_line_is_searchable_again() -> None: + """The exact shape the defect report carried out of a running service.""" + line = "run 01a0b263-b0b7-71f5-bfb2-3b2aa0f1c2d3 started" + assert default_redactor().redact(line) == line + assert "" not in default_redactor().redact(line) + + +@pytest.mark.parametrize("make", [lambda: str(uuid.uuid4()), uuid7], ids=["uuid4", "uuid7"]) +def test_no_generated_uuid_is_ever_mangled(make) -> None: # type: ignore[no-untyped-def] + """A sweep, not a handful of examples. + + The old rule mangled about a quarter of all uuids, so a few hand-picked ids + would pass by luck. Only a sweep pins the behaviour. + """ + redactor = default_redactor() + for _ in range(3000): + line = f"run {make()} finished" + assert redactor.redact(line) == line + + +def test_trace_and_span_ids_survive_in_the_message_text() -> None: + """``_NEVER_REDACT`` covers structlog FIELD keys; this is the message half.""" + line = "trace_id=4bf92f3577b34da6a3ce929d0e0e4736 span_id=00f067aa0ba902b7 handled" + assert default_redactor().redact(line) == line + + +def test_a_uuid_and_a_phone_number_on_one_line() -> None: + line = "run 550e8400-e29b-41d4-a716-446655440000 caller +34911234567" + out = default_redactor().redact(line) + assert "550e8400-e29b-41d4-a716-446655440000" in out + assert "+34911234567" not in out + assert "" in out + + +def test_a_real_phone_number_still_leaves_the_log_line() -> None: + """The filter attached to every handler, not just the engine underneath it.""" + redactor = default_redactor() + log_filter = RedactionFilter(redactor, DEFAULT_PROPS.allow_fields, DEFAULT_PROPS.deny_fields) + record = logging.LogRecord( + name="app", + level=logging.INFO, + pathname=__file__, + lineno=1, + msg="run %s called %s", + args=("550e8400-e29b-41d4-a716-446655440000", "+34911234567"), + exc_info=None, + ) + assert log_filter.filter(record) is True + rendered = record.getMessage() + assert "+34911234567" not in rendered + assert "" in rendered + assert "550e8400-e29b-41d4-a716-446655440000" in rendered + + +def test_a_card_number_is_not_hidden_by_the_span_id_shape() -> None: + """A bare 16-digit card is 16 hex characters too — the letter requirement is load-bearing.""" + assert protected_spans("card 4111111111111111 end", list(PRESERVE_PATTERNS.values())) == [] + assert "4111111111111111" not in default_redactor().redact("card 4111111111111111 end") + + +def test_a_random_16_digit_run_is_still_not_a_card() -> None: + """``VALIDATORS['CREDIT_CARD']`` must keep deciding — the Luhn gate has not moved.""" + line = "order 1234567890123456 placed" + assert default_redactor().redact(line) == line + + +class TestPhonePattern: + """The rule itself, at the level the report argued about.""" + + def test_compact_e164_matches(self) -> None: + for number in ("+34911234567", "+442071838750", "+8613800138000"): + assert BUILTIN_PATTERNS["PHONE"].search(number), number + + def test_the_guards_are_word_boundaries_not_digit_boundaries(self) -> None: + # A hyphen and a hex letter both satisfy the old (? None: + # Deliberate: an unseparated 7-digit run is an id or a count far more + # often than a phone number, and the separated forms still match. + assert not BUILTIN_PATTERNS["PHONE"].search("rows 5550143 done") + assert BUILTIN_PATTERNS["PHONE"].search("call 555-0143 now") + + +class TestPreservePatterns: + """``pyfly.logging.redaction.preserve-patterns`` — the consumer's own id shapes.""" + + def test_a_consumer_pattern_is_protected(self) -> None: + # Without the allow-list this reads as a phone number and is destroyed. + naive = RegexRedactor(["PHONE"], preserve_patterns={}) + assert naive.redact("ref ORD-555-0143 shipped") != "ref ORD-555-0143 shipped" + + redactor = RegexRedactor(["PHONE"], preserve_patterns={"ORDER_REF": r"ORD-\d{3}-\d{4}"}) + assert redactor.redact("ref ORD-555-0143 shipped") == "ref ORD-555-0143 shipped" + + def test_an_invalid_preserve_pattern_is_warned_and_ignored(self, caplog: pytest.LogCaptureFixture) -> None: + with caplog.at_level(logging.WARNING, logger="pyfly.logging"): + redactor = RegexRedactor(["PHONE"], preserve_patterns={"BROKEN": "([unclosed"}) + assert "BROKEN" in caplog.text + # The redactor is still usable, and the built-in shapes still apply. + assert redactor.redact("call +34911234567") == "call " + assert redactor.redact("run 550e8400-e29b-41d4-a716-446655440000") == ( + "run 550e8400-e29b-41d4-a716-446655440000" + ) + + def test_the_property_reaches_the_built_redactor(self) -> None: + props = RedactionProperties(engine="regex", preserve_patterns={"ORDER_REF": r"ORD-\d{3}-\d{4}"}) + redactor = build_redactor(props) + assert redactor is not None + assert redactor.redact("ref ORD-555-0143") == "ref ORD-555-0143" + + +class TestProtectedSpanSplicing: + """Redaction must never crash logging, whatever the spans look like.""" + + PATTERNS = [re.compile(r"ID-\d+")] + + def test_span_at_index_zero(self) -> None: + assert protected_spans("ID-1 tail", self.PATTERNS) == [(0, 4)] + assert RegexRedactor(["PHONE"], preserve_patterns={"X": r"ID-\d+"}).redact("ID-1 555-0143") == "ID-1 " + + def test_span_at_end_of_string(self) -> None: + redactor = RegexRedactor(["PHONE"], preserve_patterns={"X": r"ID-\d+"}) + assert redactor.redact("555-0143 ID-1") == " ID-1" + + def test_two_adjacent_spans_are_merged(self) -> None: + spans = protected_spans("ID-1ID-2", self.PATTERNS) + assert spans == [(0, 8)] + + def test_overlapping_spans_from_different_patterns_are_merged(self) -> None: + patterns = [re.compile(r"abc\d"), re.compile(r"c\d\d")] + assert protected_spans("xabc12y", patterns) == [(1, 6)] + + def test_whole_string_protected(self) -> None: + redactor = RegexRedactor(["PHONE"], preserve_patterns={"X": r"^555-0143$"}) + assert redactor.redact("555-0143") == "555-0143" + + def test_empty_and_non_string_input(self) -> None: + redactor = default_redactor() + assert redactor.redact("") == "" + assert redactor.redact(None) is None # type: ignore[arg-type] diff --git a/tests/logging/test_redaction_processor.py b/tests/logging/test_redaction_processor.py index a2e320cb..63e78404 100644 --- a/tests/logging/test_redaction_processor.py +++ b/tests/logging/test_redaction_processor.py @@ -37,11 +37,14 @@ def test_structlog_allow_fields_limits_scanning(): def test_trace_ids_are_never_redacted_even_with_phone_pattern(): - # A trace id with a 7-digit island bounded by hex letters matches the PHONE pattern; - # left unguarded the redactor would mutilate it and silently break log<->trace correlation. + # A trace id with a 7-digit island bounded by hex letters used to match the PHONE + # pattern, which silently broke log<->trace correlation. Since 26.09.07 the engine + # protects the identifier shape itself, so the id survives even in a message body; + # the field guard below stays as the second line of defence, for a key whose VALUE + # is not one of the recognised identifier shapes. tid = "a1234567ffffffffffffffffffffffff" r = RegexRedactor(["PHONE"]) - assert r.redact(tid) != tid # sanity: PHONE *would* corrupt it if unguarded + assert r.redact(tid) == tid # the engine no longer mutilates it # deny_fields lists trace_id on purpose: NEVER_REDACT must win even over an explicit deny. proc = make_structlog_redactor(r, allow_fields=[], deny_fields=["trace_id"]) out = proc( diff --git a/uv.lock b/uv.lock index a8bea8a4..6d4b1d08 100644 --- a/uv.lock +++ b/uv.lock @@ -2316,7 +2316,7 @@ wheels = [ [[package]] name = "pyfly" -version = "26.9.6" +version = "26.9.7" source = { editable = "." } dependencies = [ { name = "pydantic" }, diff --git a/web/index.html b/web/index.html index 09851440..dc8c2623 100644 --- a/web/index.html +++ b/web/index.html @@ -91,7 +91,7 @@

Apache 2.0 mypy strict async-first - v26.09.06 + v26.09.07