Give SSLCredentials a usable default TLS purpose - #768
Closed
wbarnha wants to merge 11 commits into
Closed
Conversation
`SSLCredentials.__init__` defaulted `purpose` to None and passed it straight
into `ssl.create_default_context(purpose=...)`, which begins
if not isinstance(purpose, _ASN1Object):
raise TypeError(purpose)
So `SSLCredentials()` and `SSLCredentials(cafile=...)` -- any call that does not
supply an explicit `context` -- raised `TypeError: None`. The class could only
ever be constructed by handing it a context built elsewhere, which defeats the
cafile/capath/cadata parameters entirely.
Default to `ssl.Purpose.SERVER_AUTH`: the same default `create_default_context()`
itself applies, and the correct one for a client verifying a broker. It implies
`check_hostname=True` and `verify_mode=CERT_REQUIRED`, which the tests assert.
An explicitly passed `purpose` is still forwarded unchanged.
Found by the type checker in #758 and marked `XXX` there.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`VersionInfo` declares `major`, `minor` and `micro` as ints, but the module
splatted the regex groups `(prefix, version, suffix)` into them positionally:
VersionInfo(major=None, minor='0.11.5', micro='')
So `faust.version_info.major` was the `'v'` prefix or None, `.minor` was the
whole version string, `.micro` was the suffix, and `.releaselevel` was always
None -- every field wrong except by accident.
Parse the dotted numbers properly instead, putting any non-numeric tail
(`dev1+g1234`, `rc1`, a local segment) into `releaselevel`. Missing components
pad with zero, and an unparsable version degrades to `VersionInfo(0, 0, 0, ...)`
rather than raising, so `import faust` can never fail on the version string --
the old `RuntimeError('THIS IS A BROKEN RELEASE!')` branch is gone with it.
This changes a public value, which is why #758 left it marked `XXX` rather than
fixing it: code reading `faust.version_info.minor` as the version *string* must
switch to `faust.__version__`, which is unchanged. Nothing in the repo consumes
it and no docs reference it.
`_parse_version` is injected into the lazy module's `__dict__` so it is
reachable for testing; it is private and stays out of `__all__` and `dir()`.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`Collection._del_old_keys` read `_partition_timestamp_keys` with
self._partition_timestamp_keys.get((partition, window_range))
where `window_range` is the whole `(start, end)` tuple. The map is keyed
`(partition, range_end)` -- an `(int, float)` pair -- written that way by
`_maybe_set_key_ttl` and read that way by `_maybe_del_key_ttl`.
So the lookup could never hit. `triggered_windows` was always `[None, ...]`,
`window_data` stayed empty, and `on_window_close` was only ever handed the raw
per-key value instead of the aggregated window data it exists to receive.
Use `(partition, window_range[1])`, matching the writer.
This is user-visible: applications with an `on_window_close` handler will start
receiving the aggregated data the API always promised. That is why #758 marked
it `XXX` instead of fixing it.
Two existing tests relied on `mock_ranges` returning bare floats, which is not
what `_window_ranges` yields; they now pass real `(start, end)` tuples. Their
assertions are unchanged and the ranges still match nothing in
`_partition_timestamp_keys`, so they continue to cover the untriggered path.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
Two bugs in `Recovery`, both found by the type checker in #758. `_slurp_changelogs` classifies each event's TP as active or standby and binds `table`, `offsets` and `bufsize` accordingly. The `else:` branch for a TP that is neither only logged `"recovery unknown topic"` and fell through -- so the event was applied using the *previous* iteration's bindings: written into an unrelated table's buffer and offset map, and passed to that table's `on_changelog_event`. On the first event of the loop there is nothing bound yet, so it raised `UnboundLocalError` instead. Skip applying an event for an untracked TP. Note a bare `continue` would be wrong: the statements at the bottom of the loop body -- `_maybe_signal_recovery_end()` and the standby-ready bookkeeping -- must keep running on every iteration, or recovery-end signalling loses a trigger. Only the event-application block is skipped. `detect_aborted_tx` compared `await self.app.consumer.position(tp) >= highwater` unguarded. `ConsumerT.position` is `Optional[int]` and does return None when a partition has no position yet, so that raised TypeError -- swallowed by the caller's `except Exception`, which then silently skipped the aborted-transaction fixup for every remaining partition in the loop. Skip a TP with no position. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`_app_from_str` returns None for a `require_app = False` command invoked
without `-A` -- that is the documented escape hatch, and `faust completion` is
exactly such a command. `_finalize_app` handed that None straight back, and
`AppCommand.__init__` then did
self.key_serializer = key_serializer or self.app.conf.key_serializer
unconditionally, so the command died with
`AttributeError: 'NoneType' object has no attribute 'conf'`. The escape hatch
was unusable: `faust completion` could not run without the `-A` it is written
not to need.
Make `AppCommand` tolerate having no app. `self.app` becomes a property over
an `Optional[AppT]`, the serializer defaults fall back to None when there is no
app, and `on_stop` and `blocking_timeout` no longer assume one. Behaviour with
an app present is unchanged, and a command with `require_app = True` still gets
the same `UsageError` from `_app_from_str` as before.
Found by the type checker in #758 and marked `XXX` there.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`Producer.key_partition` called `self._producer_thread.producer.list_topics()`. `ProducerThread.producer` is the *Faust* Producer; the confluent_kafka handle is `ProducerThread._producer`. Faust producers have no `list_topics`, so the method raised AttributeError and was dead on arrival. Read the confluent producer instead. `Consumer.verify_event_path` delegated to `self._thread.verify_event_path(...)`, but neither `ConsumerThread` nor `ConfluentConsumerThread` defines it, so the commit-livelock detector (`_commit_livelock_detector` -> `verify_all_partitions_active`) raised AttributeError on every tick. Add the no-op to `ConfluentConsumerThread`, matching the documented no-op stub the base `faust.transport.consumer.Consumer.verify_event_path` already is. This makes livelock detection inert for this driver rather than raising -- a real implementation is separate work. Both were found by the type checker in #758 and marked `XXX` there. Note these tests do not run here or in CI: tests/unit/transport/drivers/test_confluent.py starts with `pytest.importorskip("confluent_kafka")`, and confluent-kafka is the optional `faust[ckafka]` extra, which the CI test environment does not install. The fixes are verified by reading the class definitions, not by an executed test. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`_parse_version` carried two regexes and a hand-written loop to pull major/minor/micro off `__version__`. That is a PEP 440 parser, and there is no reason for this repo to maintain one: * the stdlib has no version parser -- `importlib.metadata.version()` returns the string only, which is why the parsing existed in the first place; * `distutils.version.LooseVersion`, the historical answer, was removed from the stdlib in Python 3.12, and this package supports 3.10 through 3.14; * `packaging.version.Version` is the PyPA reference implementation, the same parser pip and setuptools use. `packaging` is not a new install for anyone: aiokafka, a core dependency, already requires it unconditionally. It was only listed in `requirements/dist.txt` though, so it is added to `requirements.txt` -- faust imports it directly now and must not rely on a transitive dependency staying put. The floor is 20.0, where `Version.major`/`.minor`/`.micro` landed. Behaviour changes, all in the direction of the field names: * A pre/dev/post segment is now split across `releaselevel` and `serial` instead of being concatenated into `releaselevel`. `0.11.5rc1` gives `releaselevel='rc', serial='1'`; it gave `releaselevel='rc1', serial=None`. `VersionInfo` mirrors `sys.version_info`, where those two fields mean exactly this, and `serial` was previously never populated at all. * A local segment and any fourth component are dropped, because `VersionInfo` has no field for either. `0.11.5.dev1+g1234` gives `VersionInfo(0, 11, 5, 'dev', '1')`; the `+g1234` is still available in full on `faust.__version__`, which is untouched. * A string PEP 440 cannot parse degrades to `VersionInfo(0, 0, 0)` carrying the raw string, so `import faust` still cannot fail on a bad version -- same guarantee as before, now via `except InvalidVersion`. One edge case moves: `1.x.3` gave `VersionInfo(1, 0, 0, 'x.3')` and now gives `VersionInfo(0, 0, 0, '1.x.3')`, since packaging rejects it outright rather than salvaging the leading number. Folded into this PR rather than sent separately because #760 already changes `version_info` from a broken value to a correct one; doing both at once moves the public shape once instead of twice. Net: -2 regexes, -18 lines, and the `re` import drops out of `faust/__init__.py`. Verified: `mypy -p faust` clean (packaging ships py.typed, so these are real types rather than Any), pinned flake8/isort/black clean, suites 2234 -> 2237 passed, 4 skipped. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
Codecov flagged the one line of this PR's diff that no test reaches: `faust/cli/base.py:874`, the body of the `app` setter. It matters more than a coverage percentage suggests. `app` was a plain writable attribute until this PR turned it into a property over an `Optional[AppT]`; the setter exists purely so `command.app = ...` keeps working for anything that assigned to it. Nothing else in the suite exercises it, so a later refactor could drop the setter and turn every such assignment into `AttributeError` with the tests still green. `test_app__raises_when_missing` already covers the getter's raise path; this covers the round trip -- assignment lands on `_app`, and the property reads it back. faust/cli/base.py patch coverage 96% -> 100%; the file's remaining misses (86, 480-489) are pre-existing and outside this PR's diff. Suites 2237 -> 2238 passed, 4 skipped; `mypy -p faust` clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
`_del_old_keys` aggregates the data of every window overlapping an expiring timestamp and hands the result to `on_window_close`. Correcting the `_partition_timestamp_keys` lookup in c87ee2d made that block reachable for the first time, and it assumed every window value is a list: window_data.setdefault(processed_window[0], []).extend( self.data.get(processed_window, []) ) For a counting table -- `app.Table("counts", default=int).hopping(...)`, the shape the docs use -- the stored value is an `int`, and `list.extend(2)` raises `TypeError: 'int' object is not iterable`. That kills the `_clean_data` background task, so the crash is not confined to applications that define an `on_window_close` handler: `window_data` is built before any callback dispatch. A string is worse than a crash: it is iterable, so `"BOO"` and `"MOO"` across two overlapping windows silently aggregate to `['B', 'O', 'O', 'M', 'O', 'O']`. Aggregate only list values, which is the shape this exists to accumulate and the only one that concatenates without inventing semantics. Anything else is left out of `window_data` and delivered as the raw per-key value below -- exactly what handlers received for as long as the lookup was broken, so no scalar table changes behaviour. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HpEMQEwKeEfwjVPEi9cWh4
f3b2097 made `Producer.key_partition` reach the real confluent handle, so the partition arithmetic underneath it runs for the first time: key_bytes = str(key).encode("utf-8") partition = abs(hash(key_bytes)) % partition_count Two bugs in three lines. `hash()` is salted per process for bytes, so the same key resolves to a different partition in every worker and after every restart. Verified: `abs(hash(str(b"key").encode())) % 2` gives 1, 0, 0 under PYTHONHASHSEED 0, 1 and 12345. `key_partition` is what `PartitionAssignor.key_store` uses to route a request for a key to the worker owning that key's partition, and what `Channel.send` uses under `eager_partitioning`; an unstable answer sends both to the wrong place. And `str(key)` on a bytes key renders its repr, so `b"key"` is hashed as the eight bytes of `"b'key'"` rather than the three the broker sees. Replace both with `partition_for_key`, `crc32(key) % partition_count`. That is librdkafka's default `consistent_random` partitioner, which this driver never overrides, so a computed partition now matches where a keyed record produced through it actually lands. Deliberately not the Java client's murmur2 -- that is what aiokafka's `DefaultPartitioner` uses, and each driver has to mirror the library that will do its producing. `ConfluentConsumerThread.key_partition` carried the identical arithmetic and is fixed with it; leaving it would have the driver's consumer and producer disagree about where a key belongs. The tests asserted only `0 <= partition < n`, which any hash function passes, salted or not. They now assert the fixed partition crc32 picks, plus vectors for the helper and a cross-process check under three different PYTHONHASHSEEDs. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HpEMQEwKeEfwjVPEi9cWh4
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #768 +/- ##
==========================================
+ Coverage 96.05% 96.10% +0.04%
==========================================
Files 103 103
Lines 11081 11110 +29
Branches 1189 1198 +9
==========================================
+ Hits 10644 10677 +33
+ Misses 345 343 -2
+ Partials 92 90 -2 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Member
Author
|
Closing as a duplicate of #760. This was opened automatically from a working branch that shares #760's head commit ( Generated by Claude Code |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
SSLCredentials.__init__defaultedpurposeto None and passed it straightinto
ssl.create_default_context(purpose=...), which beginsSo
SSLCredentials()andSSLCredentials(cafile=...)-- any call that does notsupply an explicit
context-- raisedTypeError: None. The class could onlyever be constructed by handing it a context built elsewhere, which defeats the
cafile/capath/cadata parameters entirely.
Default to
ssl.Purpose.SERVER_AUTH: the same defaultcreate_default_context()itself applies, and the correct one for a client verifying a broker. It implies
check_hostname=Trueandverify_mode=CERT_REQUIRED, which the tests assert.An explicitly passed
purposeis still forwarded unchanged.Found by the type checker in #758 and marked
XXXthere.Co-Authored-By: Claude Opus 5 noreply@anthropic.com
Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7