fix(concurrency): handle simultaneous (a)finalize calls

[rmlibre]

Is this a new bug in aiootp?

• This affects a supported version of aiootp
• This isn't a security vulnerability or issue
• I've searched the existing issues, & I couldn't find one for this bug

Issue

Cipher streaming object (a)finalize calls can be made simultaneously in concurrent contexts, leading to invalid data processing.

Remediation

Utilize ConcurrencyGuard and atomic deque objects to track and allow the first call to (a)finalize while raising IncoherentConcurrencyState for all other calls.

aiootp/aiootp/ciphers/cipher_streams.py

Lines 193 to 209 in 1bdbfba

	self._finalizing_now.append(token := token_bytes(32))
	if not compare_digest(token, self._finalizing_now[0]):
	raise ConcurrencyGuard.IncoherentConcurrencyState
	async with ConcurrencyGuard(self._digesting_now, token=token):
	end_padding = await self._padding.aend_padding(self._byte_count)
	final_blocks = abatch(
	self._buffer.pop() + end_padding,
	size=self._config.BLOCKSIZE,
	)
	async for block in final_blocks:
	self._buffer.append(block)
	while self._buffer:
	block = await self._stream.asend(None)
	block_id = await self.shmac.anext_block_id(block)
	yield block_id, block
	await self.shmac.afinalize()

aiootp/aiootp/ciphers/decipher_streams.py

Lines 251 to 266 in 1bdbfba

	self._finalizing_now.append(token := token_bytes(32))
	if not compare_digest(token, self._finalizing_now[0]):
	raise ConcurrencyGuard.IncoherentConcurrencyState
	async with ConcurrencyGuard(self._digesting_now, token=token):
	await self.shmac.afinalize()
	async for result in self:
	yield result
	queue = self._result_queue
	footer_index = await self._padding.adepadding_end_index(
	queue[-1]
	)
	async for block in abatch(
	b"".join(queue)[:footer_index], size=self._config.BLOCKSIZE
	):
	yield block

---

The buffers that are queued to run before the (a)finalize call are allowed to finish, but all buffers queued to run afterwards raise CipherStreamIsClosed.

aiootp/aiootp/ciphers/cipher_streams.py

Lines 255 to 260 in df07919

	async with ConcurrencyGuard(self._digesting_now):
	if (
	self._finalizing_now
	and self._finalizing_now[0] not in self._digesting_now
	):
	raise CipherStreamIssue.stream_has_been_closed()

aiootp/aiootp/ciphers/decipher_streams.py

Lines 320 to 325 in df07919

	async with ConcurrencyGuard(self._digesting_now):
	if (
	self._finalizing_now
	and self._finalizing_now[0] not in self._digesting_now
	):
	raise CipherStreamIssue.stream_has_been_closed()

---

Tests were written to ensure the correct exception is raised.

Environment

- OS: Any
- Python: >=3.8
- aiootp: < 0.23.11

Additional Context

Refs (bd5cd12)

[rmlibre]

Resolved by (1bdbfba, 3d70ea9)