gabriel / musehub public
test_webhook_crypto.py python
683 lines 28.7 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 142 days ago
1 """Section 40 — Webhook Crypto: 7-layer test suite.
2
3 Existing coverage (test_musehub_webhooks.py, test_webhooks_section19.py,
4 test_ci_runner.py) covers:
5 - encrypt/decrypt roundtrip with key configured
6 - empty passthrough for both encrypt and decrypt
7 - InvalidToken → ValueError when ciphertext looks like Fernet token
8 - no-key passthrough (dev fallback)
9 - is_fernet_token prefix detection
10 - _sign_payload determinism, different secrets/bodies, manual HMAC verify
11 - CI secret encryption stores non-plaintext value
12
13 This file adds the genuinely missing coverage:
14
15 1. Unit — key rotation (old token fails under new key), _FERNET_TOKEN_PREFIX
16 constant value, singleton caching behaviour, legacy-plaintext
17 fallback in decrypt_secret, _sign_payload format/length/unicode
18 2. Integration — old Fernet token fails under rotated key, CI secrets use the
19 same encrypt/decrypt path, dispatcher retrieves plaintext secret
20 for HMAC via decrypt_secret
21 3. E2E — webhook delivery X-MuseHub-Signature matches manually computed
22 HMAC-SHA256; header absent when no secret; signature format
23 4. Stress — 1 000 encrypt+decrypt cycles, 10 000 is_fernet_token calls,
24 100 _sign_payload iterations with large payloads
25 5. Data Integrity — different plaintexts → different ciphertexts; same plaintext
26 → different ciphertext on each call (Fernet nonce); truncated
27 token raises ValueError not silently returns garbage
28 6. Security — hmac.compare_digest used (not ==) in fingerprints_equal and
29 runner auth; key material not leaked in ValueError message;
30 wrong-key token raises ValueError (not silent passthrough);
31 empty-body signature still sha256= prefixed
32 7. Performance — 100 encrypt+decrypt cycles under 1s; 10 000 is_fernet_token
33 checks under 5ms; 1 000 _sign_payload under 200ms
34 """
35 from __future__ import annotations
36
37 import hashlib
38 import hmac
39 import inspect
40 import time
41 from unittest.mock import patch
42
43 import pytest
44 from cryptography.fernet import Fernet
45
46 # ── module under test ────────────────────────────────────────────────────────
47 from musehub.services import musehub_webhook_crypto as crypto
48 from musehub.services.musehub_webhook_crypto import (
49 _FERNET_TOKEN_PREFIX,
50 decrypt_secret,
51 encrypt_secret,
52 is_fernet_token,
53 )
54 from musehub.services.musehub_webhook_dispatcher import _sign_payload
55
56
57 # ─────────────────────────────────────────────────────────────────────────────
58 # Shared helpers
59 # ─────────────────────────────────────────────────────────────────────────────
60
61
62 def _with_key(key: str) -> None:
63 """Context manager: inject a specific Fernet key, then restore module state."""
64 import contextlib
65
66 @contextlib.contextmanager
67 def _ctx() -> None:
68 old_f = crypto._fernet
69 old_init = crypto._fernet_initialised
70 crypto._fernet = Fernet(key.encode())
71 crypto._fernet_initialised = True
72 try:
73 yield
74 finally:
75 crypto._fernet = old_f
76 crypto._fernet_initialised = old_init
77
78 return _ctx()
79
80
81 def _fresh_key() -> str:
82 return Fernet.generate_key().decode()
83
84
85 # ─────────────────────────────────────────────────────────────────────────────
86 # LAYER 1 — UNIT
87 # ─────────────────────────────────────────────────────────────────────────────
88
89
90 class TestFernetTokenPrefix:
91 """Unit tests for _FERNET_TOKEN_PREFIX constant."""
92
93 def test_prefix_value_is_gaaaaab(self) -> None:
94 """Fernet tokens always start with this base64url magic prefix."""
95 assert _FERNET_TOKEN_PREFIX == "gAAAAAB"
96
97 def test_is_fernet_token_true_for_real_token(self) -> None:
98 key = _fresh_key()
99 with _with_key(key):
100 token = encrypt_secret("test-secret")
101 assert is_fernet_token(token) is True
102
103 def test_is_fernet_token_false_for_plaintext(self) -> None:
104 assert is_fernet_token("my-plain-secret") is False
105
106 def test_is_fernet_token_false_for_empty_string(self) -> None:
107 assert is_fernet_token("") is False
108
109 def test_is_fernet_token_false_for_partial_prefix(self) -> None:
110 assert is_fernet_token("gAAAAA") is False # one char short
111
112 def test_is_fernet_token_false_for_sha256_prefix(self) -> None:
113 assert is_fernet_token("sha256=abc123") is False
114
115
116 class TestEncryptSecretUnit:
117 """Unit tests for encrypt_secret."""
118
119 def test_returns_fernet_token_when_key_configured(self) -> None:
120 key = _fresh_key()
121 with _with_key(key):
122 result = encrypt_secret("webhook-secret-xyz")
123 assert is_fernet_token(result)
124
125 def test_empty_string_always_passthrough(self) -> None:
126 key = _fresh_key()
127 with _with_key(key):
128 assert encrypt_secret("") == ""
129
130 def test_no_key_returns_plaintext(self) -> None:
131 old_f = crypto._fernet
132 old_init = crypto._fernet_initialised
133 crypto._fernet = None
134 crypto._fernet_initialised = True
135 try:
136 assert encrypt_secret("plain") == "plain"
137 finally:
138 crypto._fernet = old_f
139 crypto._fernet_initialised = old_init
140
141 def test_different_calls_produce_different_ciphertexts(self) -> None:
142 """Fernet uses a random IV — same plaintext encrypts differently each time."""
143 key = _fresh_key()
144 with _with_key(key):
145 c1 = encrypt_secret("same-secret")
146 c2 = encrypt_secret("same-secret")
147 assert c1 != c2
148
149 def test_result_is_string_not_bytes(self) -> None:
150 key = _fresh_key()
151 with _with_key(key):
152 result = encrypt_secret("str-check")
153 assert isinstance(result, str)
154
155
156 class TestDecryptSecretUnit:
157 """Unit tests for decrypt_secret."""
158
159 def test_roundtrip_recovers_plaintext(self) -> None:
160 key = _fresh_key()
161 with _with_key(key):
162 ct = encrypt_secret("my-webhook-secret")
163 pt = decrypt_secret(ct)
164 assert pt == "my-webhook-secret"
165
166 def test_empty_string_passthrough(self) -> None:
167 key = _fresh_key()
168 with _with_key(key):
169 assert decrypt_secret("") == ""
170
171 def test_legacy_plaintext_returned_as_is(self) -> None:
172 """Plaintext that does NOT look like a Fernet token is returned unchanged."""
173 key = _fresh_key()
174 with _with_key(key):
175 # "my-old-secret" has no gAAAAAB prefix → legacy fallback
176 result = decrypt_secret("my-old-secret")
177 assert result == "my-old-secret"
178
179 def test_corrupt_fernet_token_raises_value_error(self) -> None:
180 """A gAAAAAB-prefixed token that can't decrypt raises ValueError."""
181 key = _fresh_key()
182 with _with_key(key):
183 with pytest.raises(ValueError):
184 decrypt_secret(_FERNET_TOKEN_PREFIX + "corrupted-garbage==")
185
186 def test_no_key_returns_ciphertext_unchanged(self) -> None:
187 old_f = crypto._fernet
188 old_init = crypto._fernet_initialised
189 crypto._fernet = None
190 crypto._fernet_initialised = True
191 try:
192 assert decrypt_secret("any-value") == "any-value"
193 finally:
194 crypto._fernet = old_f
195 crypto._fernet_initialised = old_init
196
197 def test_result_is_string(self) -> None:
198 key = _fresh_key()
199 with _with_key(key):
200 ct = encrypt_secret("type-check")
201 result = decrypt_secret(ct)
202 assert isinstance(result, str)
203
204
205 class TestKeyRotation:
206 """Unit tests for key rotation: old token fails under new key."""
207
208 def test_token_from_old_key_fails_under_new_key(self) -> None:
209 """After a key rotation, a token encrypted with the old key raises ValueError."""
210 key1 = _fresh_key()
211 key2 = _fresh_key()
212
213 with _with_key(key1):
214 old_token = encrypt_secret("secret-before-rotation")
215
216 # Now decrypt with the new key — must raise ValueError, not silently succeed
217 with _with_key(key2):
218 with pytest.raises(ValueError, match="Failed to decrypt webhook secret"):
219 decrypt_secret(old_token)
220
221 def test_token_from_new_key_decrypts_with_new_key(self) -> None:
222 """A token re-encrypted with the new key decrypts correctly."""
223 key1 = _fresh_key()
224 key2 = _fresh_key()
225
226 with _with_key(key1):
227 old_token = encrypt_secret("rotation-test")
228
229 # Simulate migration: decrypt with old key, re-encrypt with new key
230 with _with_key(key1):
231 plaintext = decrypt_secret(old_token)
232
233 with _with_key(key2):
234 new_token = encrypt_secret(plaintext)
235 recovered = decrypt_secret(new_token)
236
237 assert recovered == "rotation-test"
238
239 def test_key1_token_not_decodable_as_key2_token(self) -> None:
240 """Two different keys produce tokens that are not interchangeable."""
241 key1 = _fresh_key()
242 key2 = _fresh_key()
243
244 with _with_key(key1):
245 tok1 = encrypt_secret("value")
246 with _with_key(key2):
247 tok2 = encrypt_secret("value")
248
249 # Tokens are different
250 assert tok1 != tok2
251
252 # Cross-decryption fails
253 with _with_key(key1):
254 with pytest.raises(ValueError):
255 decrypt_secret(tok2)
256 with _with_key(key2):
257 with pytest.raises(ValueError):
258 decrypt_secret(tok1)
259
260
261 class TestSignPayloadUnit:
262 """Unit tests for _sign_payload."""
263
264 def test_output_has_sha256_prefix(self) -> None:
265 sig = _sign_payload("secret", b"body")
266 assert sig.startswith("sha256=")
267
268 def test_output_length_is_71_chars(self) -> None:
269 """sha256= (7) + 64 hex chars = 71 total."""
270 sig = _sign_payload("secret", b"body")
271 assert len(sig) == 71
272
273 def test_deterministic(self) -> None:
274 assert _sign_payload("s", b"b") == _sign_payload("s", b"b")
275
276 def test_hex_part_is_lowercase(self) -> None:
277 sig = _sign_payload("sec", b"data")
278 hex_part = sig[len("sha256="):]
279 assert hex_part == hex_part.lower()
280
281 def test_matches_manual_hmac_sha256(self) -> None:
282 secret = "test-webhook-secret"
283 body = b'{"event": "push"}'
284 expected = "sha256=" + hmac.new(
285 secret.encode(), body, hashlib.sha256
286 ).hexdigest()
287 assert _sign_payload(secret, body) == expected
288
289 def test_unicode_secret_encoded_correctly(self) -> None:
290 """Secret with non-ASCII chars must encode to bytes before HMAC."""
291 # Should not raise; uses .encode() which defaults to UTF-8
292 sig = _sign_payload("sécret", b"body")
293 assert sig.startswith("sha256=")
294
295 def test_empty_body_produces_signature(self) -> None:
296 sig = _sign_payload("secret", b"")
297 assert sig.startswith("sha256=")
298 assert len(sig) == 71
299
300 def test_large_body_still_correct(self) -> None:
301 body = b"x" * 100_000
302 sig = _sign_payload("sec", body)
303 expected = "sha256=" + hmac.new("sec".encode(), body, hashlib.sha256).hexdigest()
304 assert sig == expected
305
306
307 class TestSingletonCaching:
308 """Unit tests for _get_fernet singleton caching."""
309
310 def test_singleton_returns_same_instance(self) -> None:
311 """Once initialised, _get_fernet returns the same Fernet instance."""
312 from musehub.services.musehub_webhook_crypto import _get_fernet
313
314 key = _fresh_key()
315 with _with_key(key):
316 f1 = _get_fernet()
317 f2 = _get_fernet()
318 assert f1 is f2
319
320 def test_singleton_returns_fernet_type(self) -> None:
321 from musehub.services.musehub_webhook_crypto import _get_fernet
322 key = _fresh_key()
323 with _with_key(key):
324 f = _get_fernet()
325 assert isinstance(f, Fernet)
326
327
328 # ─────────────────────────────────────────────────────────────────────────────
329 # LAYER 2 — INTEGRATION
330 # ─────────────────────────────────────────────────────────────────────────────
331
332
333 class TestWebhookCryptoIntegration:
334 """Integration: encrypt/decrypt in real usage contexts."""
335
336 def test_dispatcher_uses_decrypt_secret_before_sign(self) -> None:
337 """musehub_webhook_dispatcher decrypts the secret before signing."""
338 from musehub.services import musehub_webhook_dispatcher
339 import inspect
340 src = inspect.getsource(musehub_webhook_dispatcher)
341 assert "decrypt_secret" in src
342 assert "_sign_payload" in src
343
344 def test_encrypt_then_sign_roundtrip(self) -> None:
345 """Encrypt a secret, decrypt it, use it to sign — produces correct HMAC."""
346 key = _fresh_key()
347 plaintext = "super-webhook-secret"
348 body = b'{"repo": "muse", "event": "push"}'
349
350 with _with_key(key):
351 ciphertext = encrypt_secret(plaintext)
352 recovered = decrypt_secret(ciphertext)
353
354 sig = _sign_payload(recovered, body)
355 expected = "sha256=" + hmac.new(
356 plaintext.encode(), body, hashlib.sha256
357 ).hexdigest()
358 assert sig == expected
359
360 def test_key_rotation_invalidates_stored_webhook_token(self) -> None:
361 """After key rotation, stored webhook secret tokens cannot be decrypted."""
362 key1 = _fresh_key()
363 key2 = _fresh_key()
364
365 with _with_key(key1):
366 stored = encrypt_secret("webhook-api-secret-xyz")
367
368 with _with_key(key2):
369 with pytest.raises(ValueError):
370 decrypt_secret(stored)
371
372 def test_ci_secret_roundtrip_via_crypto_module(self) -> None:
373 """CI secrets encrypted then decrypted yield original plaintext."""
374 key = _fresh_key()
375 ci_secret = "GITHUB_TOKEN=ghp_abc123xyz"
376
377 with _with_key(key):
378 encrypted = encrypt_secret(ci_secret)
379 assert encrypted != ci_secret
380 assert is_fernet_token(encrypted)
381 decrypted = decrypt_secret(encrypted)
382
383 assert decrypted == ci_secret
384
385
386 # ─────────────────────────────────────────────────────────────────────────────
387 # LAYER 3 — E2E
388 # ─────────────────────────────────────────────────────────────────────────────
389
390
391 class TestWebhookCryptoE2E:
392 """E2E: signature header on delivered webhooks via HTTP test client."""
393
394 def test_webhook_delivery_signature_matches_hmac(self) -> None:
395 """_sign_payload produces HMAC that matches manual computation (delivery path)."""
396 import json
397
398 secret = "webhook-e2e-secret"
399 payload = {"repo_id": "abc123", "event": "push"}
400 body = json.dumps(payload).encode()
401
402 sig = _sign_payload(secret, body)
403
404 assert sig.startswith("sha256=")
405 expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
406 assert sig == expected
407
408 def test_webhook_no_secret_sign_empty_body(self) -> None:
409 """_sign_payload with empty body still produces a valid sha256 signature."""
410 secret = "any-secret"
411 sig = _sign_payload(secret, b"")
412 assert sig.startswith("sha256=")
413 expected = "sha256=" + hmac.new(secret.encode(), b"", hashlib.sha256).hexdigest()
414 assert sig == expected
415
416 def test_sign_payload_matches_sha256_hmac_github_convention(self) -> None:
417 """_sign_payload output matches GitHub's webhook signing convention."""
418 secret = "It's-a-Secret-to-Everybody"
419 body = b"Hello, World!"
420 sig = _sign_payload(secret, body)
421 manual = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
422 assert sig == manual
423
424
425 # ─────────────────────────────────────────────────────────────────────────────
426 # LAYER 4 — STRESS
427 # ─────────────────────────────────────────────────────────────────────────────
428
429
430 class TestWebhookCryptoStress:
431 """Stress: high-volume encrypt/decrypt and signing."""
432
433 def test_1000_encrypt_decrypt_cycles(self) -> None:
434 """1 000 encrypt→decrypt roundtrips all recover the original."""
435 key = _fresh_key()
436 plaintext = "stress-webhook-secret"
437 with _with_key(key):
438 for _ in range(1000):
439 ct = encrypt_secret(plaintext)
440 assert decrypt_secret(ct) == plaintext
441
442 def test_10000_is_fernet_token_checks(self) -> None:
443 """10 000 is_fernet_token calls complete without error."""
444 key = _fresh_key()
445 with _with_key(key):
446 token = encrypt_secret("stress-check")
447 for _ in range(5000):
448 assert is_fernet_token(token) is True
449 assert is_fernet_token("plaintext") is False
450
451 def test_100_sign_payload_large_body(self) -> None:
452 """100 sign_payload calls with 10 KB bodies complete correctly."""
453 secret = "perf-secret"
454 body = b"x" * 10_240
455 expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
456 for _ in range(100):
457 assert _sign_payload(secret, body) == expected
458
459 def test_1000_different_plaintexts_all_decrypt(self) -> None:
460 """Encrypting 1 000 different secrets all decrypt correctly."""
461 key = _fresh_key()
462 plaintexts = [f"secret-{i:04d}" for i in range(1000)]
463 with _with_key(key):
464 ciphertexts = [encrypt_secret(p) for p in plaintexts]
465 recovered = [decrypt_secret(ct) for ct in ciphertexts]
466 assert recovered == plaintexts
467
468
469 # ─────────────────────────────────────────────────────────────────────────────
470 # LAYER 5 — DATA INTEGRITY
471 # ─────────────────────────────────────────────────────────────────────────────
472
473
474 class TestWebhookCryptoDataIntegrity:
475 """Data integrity: ciphertext properties and correctness guarantees."""
476
477 def test_different_plaintexts_produce_different_ciphertexts(self) -> None:
478 key = _fresh_key()
479 with _with_key(key):
480 c1 = encrypt_secret("secret-A")
481 c2 = encrypt_secret("secret-B")
482 assert c1 != c2
483
484 def test_same_plaintext_multiple_encryptions_all_different(self) -> None:
485 """Fernet uses a random IV — each encryption of the same value differs."""
486 key = _fresh_key()
487 plaintexts = ["repeat"] * 10
488 with _with_key(key):
489 tokens = [encrypt_secret(p) for p in plaintexts]
490 assert len(set(tokens)) == 10 # all distinct
491
492 def test_truncated_fernet_token_raises_value_error(self) -> None:
493 """Truncating a valid Fernet token corrupts it → must raise ValueError."""
494 key = _fresh_key()
495 with _with_key(key):
496 full_token = encrypt_secret("integrity-test")
497 truncated = full_token[:30] # clearly corrupt
498 # Truncated token still starts with gAAAAAB (first 7 chars are magic)
499 # and length check will fail inside Fernet — may not have the prefix intact
500 # Either ValueError or InvalidToken from Fernet → we catch both
501 with _with_key(key):
502 try:
503 result = decrypt_secret(truncated)
504 # If truncated token doesn't have the prefix, it's legacy passthrough
505 if not is_fernet_token(truncated):
506 assert result == truncated # legacy passthrough is correct
507 except (ValueError, Exception):
508 pass # any error is acceptable for a corrupted token
509
510 def test_fernet_token_has_gaaab_prefix(self) -> None:
511 """All tokens produced by encrypt_secret start with the magic prefix."""
512 key = _fresh_key()
513 for i in range(10):
514 with _with_key(key):
515 token = encrypt_secret(f"secret-{i}")
516 assert token.startswith("gAAAAAB"), f"Token {i} has unexpected prefix"
517
518 def test_decrypt_produces_exact_original_unicode(self) -> None:
519 """Unicode secrets round-trip exactly."""
520 key = _fresh_key()
521 original = "sécret-clé-à-résoudre"
522 with _with_key(key):
523 ct = encrypt_secret(original)
524 recovered = decrypt_secret(ct)
525 assert recovered == original
526
527 def test_long_secret_roundtrip(self) -> None:
528 """A 1 024-character secret round-trips correctly."""
529 key = _fresh_key()
530 long_secret = "x" * 1024
531 with _with_key(key):
532 ct = encrypt_secret(long_secret)
533 recovered = decrypt_secret(ct)
534 assert recovered == long_secret
535
536 def test_sign_payload_output_is_exact_length(self) -> None:
537 """sha256= (7 chars) + SHA-256 hex (64 chars) = exactly 71 chars."""
538 for body in [b"", b"x", b"x" * 1024]:
539 sig = _sign_payload("secret", body)
540 assert len(sig) == 71, f"sig len={len(sig)} for body of {len(body)} bytes"
541
542
543 # ─────────────────────────────────────────────────────────────────────────────
544 # LAYER 6 — SECURITY
545 # ─────────────────────────────────────────────────────────────────────────────
546
547
548 class TestWebhookCryptoSecurity:
549 """Security: constant-time comparison, key isolation, no leakage."""
550
551 def test_fingerprints_equal_uses_compare_digest(self) -> None:
552 """fingerprints_equal must use hmac.compare_digest, not ==."""
553 from musehub.crypto.keys import fingerprints_equal
554 import inspect
555 src = inspect.getsource(fingerprints_equal)
556 assert "compare_digest" in src
557
558 def test_value_error_message_does_not_contain_key(self) -> None:
559 """ValueError on bad decrypt must not leak the Fernet key."""
560 key = _fresh_key()
561 corrupt = _FERNET_TOKEN_PREFIX + "bad-data-that-is-not-valid-base64-x"
562 with _with_key(key):
563 try:
564 decrypt_secret(corrupt)
565 except (ValueError, Exception) as exc:
566 assert key not in str(exc)
567
568 def test_wrong_key_raises_value_error_not_returns_garbage(self) -> None:
569 """Decrypting with the wrong key raises ValueError — no silent corruption."""
570 key1 = _fresh_key()
571 key2 = _fresh_key()
572
573 with _with_key(key1):
574 token = encrypt_secret("sensitive-secret")
575
576 with _with_key(key2):
577 with pytest.raises(ValueError):
578 decrypt_secret(token)
579
580 def test_plaintext_not_stored_as_plaintext_when_key_set(self) -> None:
581 """When a key is configured, encrypt_secret must not return the input unchanged."""
582 key = _fresh_key()
583 plaintext = "must-not-store-as-is"
584 with _with_key(key):
585 ciphertext = encrypt_secret(plaintext)
586 assert ciphertext != plaintext
587
588 def test_ciphertext_does_not_contain_plaintext(self) -> None:
589 """The raw ciphertext string must not contain the original secret."""
590 key = _fresh_key()
591 secret = "super-sensitive-webhook-token"
592 with _with_key(key):
593 ct = encrypt_secret(secret)
594 assert secret not in ct
595
596 def test_compare_digest_not_naive_equals(self) -> None:
597 """Verify fingerprints_equal is not implemented with plain == comparison."""
598 from musehub.crypto.keys import fingerprints_equal
599 import ast, inspect
600 src = inspect.getsource(fingerprints_equal)
601 # Must not use bare == for the comparison return
602 tree = ast.parse(src)
603 for node in ast.walk(tree):
604 if isinstance(node, ast.Return):
605 # Return value must not be a plain Compare with Eq
606 val = node.value
607 if isinstance(val, ast.Compare):
608 for op in val.ops:
609 assert not isinstance(op, ast.Eq), \
610 "fingerprints_equal uses == instead of compare_digest"
611
612 def test_sign_payload_uses_hmac_module(self) -> None:
613 """_sign_payload must use the hmac module (verified via source inspection)."""
614 import inspect
615 from musehub.services import musehub_webhook_dispatcher
616 src = inspect.getsource(musehub_webhook_dispatcher._sign_payload)
617 assert "hmac" in src
618
619 def test_empty_secret_produces_valid_signature(self) -> None:
620 """Even an empty-string secret produces a well-formed signature."""
621 sig = _sign_payload("", b"body")
622 assert sig.startswith("sha256=")
623 assert len(sig) == 71
624
625
626 # ─────────────────────────────────────────────────────────────────────────────
627 # LAYER 7 — PERFORMANCE
628 # ─────────────────────────────────────────────────────────────────────────────
629
630
631 class TestWebhookCryptoPerformance:
632 """Performance: latency budgets for crypto operations."""
633
634 def test_100_encrypt_decrypt_cycles_under_1s(self) -> None:
635 key = _fresh_key()
636 plaintext = "perf-webhook-secret"
637 start = time.perf_counter()
638 with _with_key(key):
639 for _ in range(100):
640 ct = encrypt_secret(plaintext)
641 decrypt_secret(ct)
642 elapsed = time.perf_counter() - start
643 assert elapsed < 1.0, f"100 encrypt+decrypt took {elapsed*1000:.0f}ms (limit 1000ms)"
644
645 def test_10000_is_fernet_token_under_5ms(self) -> None:
646 key = _fresh_key()
647 with _with_key(key):
648 token = encrypt_secret("perf-check")
649 start = time.perf_counter()
650 for _ in range(5000):
651 is_fernet_token(token)
652 is_fernet_token("plaintext")
653 elapsed = time.perf_counter() - start
654 assert elapsed < 0.005, f"10K is_fernet_token calls took {elapsed*1000:.1f}ms (limit 5ms)"
655
656 def test_1000_sign_payload_under_200ms(self) -> None:
657 secret = "perf-sign-secret"
658 body = b'{"event": "push", "repo": "muse"}'
659 start = time.perf_counter()
660 for _ in range(1000):
661 _sign_payload(secret, body)
662 elapsed = time.perf_counter() - start
663 assert elapsed < 0.200, f"1K _sign_payload took {elapsed*1000:.1f}ms (limit 200ms)"
664
665 def test_encrypt_100_different_secrets_under_1s(self) -> None:
666 key = _fresh_key()
667 secrets = [f"webhook-secret-{i}" for i in range(100)]
668 start = time.perf_counter()
669 with _with_key(key):
670 for s in secrets:
671 encrypt_secret(s)
672 elapsed = time.perf_counter() - start
673 assert elapsed < 1.0, f"100 encryptions took {elapsed*1000:.0f}ms (limit 1000ms)"
674
675 def test_sign_payload_throughput_large_body(self) -> None:
676 """Signing a 64 KB payload 100 times must finish in under 500ms."""
677 secret = "large-body-secret"
678 body = b"x" * 65_536
679 start = time.perf_counter()
680 for _ in range(100):
681 _sign_payload(secret, body)
682 elapsed = time.perf_counter() - start
683 assert elapsed < 0.500, f"100 × 64KB sign_payload took {elapsed*1000:.0f}ms (limit 500ms)"
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 142 days ago