From be30b8454793e0610f46c1feb35dd8e1ac30124e Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sun, 13 Sep 2026 22:56:24 -0400 Subject: [PATCH 1/4] fix(auth): eliminate sensitive token logging in verification and password recovery Remove raw token and recovery URL printing from stdout/stderr during registration and password reset workflows. Add automated test coverage confirming plaintext tokens are not leaked to server logs. --- backend/routers/auth.py | 6 +++--- tests/test_auth_api.py | 24 ++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 96e1215..679043b 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -236,7 +236,7 @@ def register(req: UserRegisterRequest, request: Request): expires_at=expires_at, ) - print(f"\n[AUTH] Generated verification link for {user.email}:\n{cfg.app_url}/verify-email?token={verify_token}\n") + print(f"[AUTH] Generated email verification link for {user.email}") try: email_service.send_verification(recipient=user.email, token=verify_token, app_url=cfg.app_url) print(f"[AUTH] Successfully dispatched verification email to {user.email}") @@ -407,14 +407,14 @@ def request_password_reset(req: PasswordResetRequest, request: Request): storage.invalidate_prior_auth_tokens(raw_user["id"], "reset_password") storage.create_auth_token(raw_user["id"], token_hash, "reset_password", expires_at) - print(f"\n[AUTH] Generated password reset link for {email}:\n{cfg.app_url}/reset-password?token={reset_token}\n") + print(f"[AUTH] Generated password reset link for {email}") try: email_service.send_password_reset(recipient=email, token=reset_token, app_url=cfg.app_url) print(f"[AUTH] Successfully dispatched password reset email to {email}") except Exception as e: print(f"[AUTH ERROR] Failed to send password reset email: {e}") else: - print(f"\n[AUTH] Password reset requested for '{email}', but no user account exists with this email address in the database.\n") + pass # Generic response to prevent account enumeration return {"message": "If an account exists with that email, a password reset link has been sent."} diff --git a/tests/test_auth_api.py b/tests/test_auth_api.py index c16f82e..d6e1b87 100644 --- a/tests/test_auth_api.py +++ b/tests/test_auth_api.py @@ -73,3 +73,27 @@ def test_auth_config_returns_client_id(): assert "google_client_id" in data assert len(data["google_client_id"]) > 0 + +def test_tokens_not_printed_to_logs(capsys): + test_email = f"logtest_{uuid.uuid4().hex[:8]}@example.com" + capsys.readouterr() + + res_reg = client.post( + "/api/auth/register", + json={"email": test_email, "password": "Password123!", "name": "Log Test User"}, + ) + assert res_reg.status_code == 200 + + out_reg, err_reg = capsys.readouterr() + assert "/verify-email?token=" not in out_reg + assert "/verify-email?token=" not in err_reg + + res_reset = client.post( + "/api/auth/password-reset/request", + json={"email": test_email}, + ) + assert res_reset.status_code == 200 + + out_reset, err_reset = capsys.readouterr() + assert "/reset-password?token=" not in out_reset + assert "/reset-password?token=" not in err_reset From 9ea75497d0c03ce5ba8f01640356cb1971474937 Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sun, 13 Sep 2026 22:57:30 -0400 Subject: [PATCH 2/4] fix(parser): enforce zip decompression limit for Word .doc files Apply the 50 MB uncompressed zip safety guard to both .docx and .doc archives before parsing with python-docx, preventing decompression bomb expansion in renamed OOXML documents. Reject non-zip legacy binary files with an informative error. --- backend/services/document_parser.py | 41 +++++++++++++---------------- tests/test_document_parser.py | 28 ++++++++++++++++++++ 2 files changed, 47 insertions(+), 22 deletions(-) diff --git a/backend/services/document_parser.py b/backend/services/document_parser.py index cc86c04..60dfef6 100644 --- a/backend/services/document_parser.py +++ b/backend/services/document_parser.py @@ -123,30 +123,27 @@ def extract_text_from_file(file_bytes: bytes, filename: str) -> str: # 2. Word (.docx, .doc) if ext in [".docx", ".doc"]: - if ext == ".doc": - # Attempt docx parse in case it was a misnamed OOXML file; otherwise reject binary .doc - try: - import docx - doc = docx.Document(io.BytesIO(file_bytes)) - except Exception: + # Guard against decompression bombs in zip-based docx and doc files + try: + with zipfile.ZipFile(io.BytesIO(file_bytes)) as zf: + total_uncompressed = sum(info.file_size for info in zf.infolist()) + if total_uncompressed > MAX_DOCX_DECOMPRESSED_BYTES: + raise ValueError( + f"Document exceeds maximum decompressed size limit of 50 MB (got {total_uncompressed // (1024*1024)} MB)." + ) + except zipfile.BadZipFile: + if ext == ".doc": raise ValueError("Legacy .doc binary format is not supported. Please convert and save your resume as .docx or .pdf.") - else: - # Guard against decompression bombs in zip-based docx files - try: - with zipfile.ZipFile(io.BytesIO(file_bytes)) as zf: - total_uncompressed = sum(info.file_size for info in zf.infolist()) - if total_uncompressed > MAX_DOCX_DECOMPRESSED_BYTES: - raise ValueError( - f"Document exceeds maximum decompressed size limit of 50 MB (got {total_uncompressed // (1024*1024)} MB)." - ) - except zipfile.BadZipFile: - pass - try: - import docx - doc = docx.Document(io.BytesIO(file_bytes)) - except Exception as e: - raise ValueError(f"Failed to read Word document: {str(e)}. Please save as .docx or .pdf.") + try: + import docx + doc = docx.Document(io.BytesIO(file_bytes)) + except ValueError: + raise + except Exception as e: + if ext == ".doc": + raise ValueError("Legacy .doc binary format is not supported. Please convert and save your resume as .docx or .pdf.") + raise ValueError(f"Failed to read Word document: {str(e)}. Please save as .docx or .pdf.") lines = [] total_chars = 0 diff --git a/tests/test_document_parser.py b/tests/test_document_parser.py index 1fb2bed..e956dbd 100644 --- a/tests/test_document_parser.py +++ b/tests/test_document_parser.py @@ -1,6 +1,7 @@ import io import docx import pypdf +import pytest from backend.services.document_parser import extract_text_from_file def test_extract_from_text(): @@ -35,3 +36,30 @@ def test_extract_from_pdf(): text = extract_text_from_file(pdf_bytes, "sample.pdf") # Even on blank page, should return string without crashing assert isinstance(text, str) + + +def test_doc_decompression_limit_blocks_renamed_bomb(): + import zipfile + bio = io.BytesIO() + with zipfile.ZipFile(bio, "w", compression=zipfile.ZIP_DEFLATED) as zf: + zf.writestr("word/document.xml", b"0" * (51 * 1024 * 1024)) + bio.seek(0) + bomb_bytes = bio.getvalue() + + with pytest.raises(ValueError, match="50 MB"): + extract_text_from_file(bomb_bytes, "bomb.doc") + + +def test_doc_renamed_valid_docx_succeeds(): + doc = docx.Document() + doc.add_heading("Alice Smith", level=1) + doc.add_paragraph("Python Developer") + bio = io.BytesIO() + doc.save(bio) + text = extract_text_from_file(bio.getvalue(), "resume.doc") + assert "Alice Smith" in text + + +def test_doc_binary_format_rejected(): + with pytest.raises(ValueError, match="Legacy .doc binary format is not supported"): + extract_text_from_file(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"random binary", "legacy.doc") From 4d61db1e8aa525ffa2be4a0925c3106744ac50ed Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sun, 13 Sep 2026 22:58:51 -0400 Subject: [PATCH 3/4] fix(storage): enforce atomic storage quota reservations for concurrent uploads Reserve account storage bytes atomically in database before uploading to object storage, releasing reservations on failure. Count reserved and active attachments towards storage limit to prevent concurrent uploads from exceeding quota. --- backend/services/resume_files.py | 45 ++++++++++------ backend/storage.py | 91 +++++++++++++++++++++++++++++++- tests/test_atomic_limits.py | 43 +++++++++++++++ 3 files changed, 161 insertions(+), 18 deletions(-) diff --git a/backend/services/resume_files.py b/backend/services/resume_files.py index 8f0a87a..b599d19 100644 --- a/backend/services/resume_files.py +++ b/backend/services/resume_files.py @@ -31,29 +31,35 @@ def reserve_and_upload( max_resumes: int = 10, max_storage_bytes: int = 50 * 1024 * 1024, ) -> Resume: - current_bytes = self.storage.get_user_upload_bytes(user_id) - if current_bytes + len(content_bytes) > max_storage_bytes: - raise ValueError("Maximum total storage limit of 50MB reached for resumes.") - - file_key = self.object_storage.upload_file( - content_bytes=content_bytes, - filename=filename, - content_type=content_type, - user_id=user_id, - ) - storage_backend = "r2" if self.object_storage.is_configured else "local" - attachment = self.storage.create_attachment( + + attachment = self.storage.reserve_upload_bytes( user_id=user_id, - storage_backend=storage_backend, - object_key=file_key, + size_bytes=len(content_bytes), + max_storage_bytes=max_storage_bytes, + max_resumes=max_resumes, original_filename=filename, content_type=content_type, - size_bytes=len(content_bytes), + storage_backend=storage_backend, ) - final_name = resume_name.strip() if (resume_name and resume_name.strip()) else filename + file_key = None try: + file_key = self.object_storage.upload_file( + content_bytes=content_bytes, + filename=filename, + content_type=content_type, + user_id=user_id, + ) + + self.storage.finalize_attachment( + attachment_id=attachment.id, + object_key=file_key, + storage_backend=storage_backend, + user_id=user_id, + ) + + final_name = resume_name.strip() if (resume_name and resume_name.strip()) else filename resume = self.storage.add_resume( name=final_name, content=text_content, @@ -64,10 +70,15 @@ def reserve_and_upload( ) return resume except Exception: - self.object_storage.delete_file(file_key, user_id=user_id) + if file_key: + try: + self.object_storage.delete_file(file_key, user_id=user_id) + except Exception: + pass self.storage.update_attachment_deletion_state(attachment.id, "failed", user_id=user_id) raise + def delete_resume(self, user_id: str, resume_id: str) -> bool: resume = self.storage.get_resume(resume_id, user_id=user_id) if not resume: diff --git a/backend/storage.py b/backend/storage.py index 8fe58a4..cff790c 100644 --- a/backend/storage.py +++ b/backend/storage.py @@ -661,7 +661,7 @@ def get_user_upload_bytes(self, user_id: str) -> int: with self._get_cursor() as cursor: cursor.execute( self._format_sql( - "SELECT COALESCE(SUM(size_bytes), 0) as total_bytes FROM attachments WHERE user_id = ? AND deletion_state = 'active'" + "SELECT COALESCE(SUM(size_bytes), 0) as total_bytes FROM attachments WHERE user_id = ? AND deletion_state IN ('active', 'reserved')" ), (user_id,), ) @@ -672,7 +672,96 @@ def get_user_upload_bytes(self, user_id: str) -> int: return int(row.get("total_bytes", 0)) return int(row[0]) + def reserve_upload_bytes( + self, + user_id: str, + size_bytes: int, + max_storage_bytes: int, + max_resumes: Optional[int] = None, + original_filename: Optional[str] = None, + content_type: Optional[str] = None, + storage_backend: str = "local", + ) -> ResumeAttachment: + now = datetime.now().isoformat() + att_id = str(uuid.uuid4()) + placeholder_key = f"reserved_{att_id}" + + with self._resume_quota_lock: + with self._get_immediate_cursor() as cursor: + if self.is_postgres: + cursor.execute("SELECT id FROM users WHERE id = %s FOR UPDATE", (user_id,)) + + if max_resumes is not None: + cursor.execute( + self._format_sql("SELECT COUNT(*) as cnt FROM resumes WHERE user_id = ?"), + (user_id,), + ) + row = cursor.fetchone() + count = int(row["cnt"]) if isinstance(row, dict) else int(row[0]) + if count >= max_resumes: + raise ValueError(f"Maximum limit of {max_resumes} resumes reached per account.") + + cursor.execute( + self._format_sql( + "SELECT COALESCE(SUM(size_bytes), 0) as total_bytes FROM attachments WHERE user_id = ? AND deletion_state IN ('active', 'reserved')" + ), + (user_id,), + ) + row = cursor.fetchone() + current_bytes = int(row["total_bytes"]) if isinstance(row, dict) else int(row[0]) + + if current_bytes + size_bytes > max_storage_bytes: + limit_mb = max_storage_bytes // (1024 * 1024) + limit_label = f"{limit_mb}MB" if limit_mb > 0 else f"{max_storage_bytes} bytes" + raise ValueError(f"Maximum total storage limit of {limit_label} reached for resumes.") + + cursor.execute( + self._format_sql( + "INSERT INTO attachments (id, user_id, storage_backend, object_key, original_filename, content_type, size_bytes, deletion_state, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ), + (att_id, user_id, storage_backend, placeholder_key, original_filename, content_type, size_bytes, "reserved", now, now), + ) + + return ResumeAttachment( + id=att_id, + user_id=user_id, + storage_backend=storage_backend, + object_key=placeholder_key, + original_filename=original_filename, + content_type=content_type, + size_bytes=size_bytes, + deletion_state="reserved", + created_at=now, + updated_at=now, + ) + + def finalize_attachment( + self, + attachment_id: str, + object_key: str, + storage_backend: str, + user_id: Optional[str] = None, + ) -> None: + now = datetime.now().isoformat() + with self._get_cursor() as cursor: + if user_id: + cursor.execute( + self._format_sql( + "UPDATE attachments SET object_key = ?, storage_backend = ?, deletion_state = 'active', updated_at = ? WHERE id = ? AND user_id = ?" + ), + (object_key, storage_backend, now, attachment_id, user_id), + ) + else: + cursor.execute( + self._format_sql( + "UPDATE attachments SET object_key = ?, storage_backend = ?, deletion_state = 'active', updated_at = ? WHERE id = ?" + ), + (object_key, storage_backend, now, attachment_id), + ) + # --- Attachments CRUD --- + def create_attachment( self, user_id: str, diff --git a/tests/test_atomic_limits.py b/tests/test_atomic_limits.py index b3d4a7a..23b7359 100644 --- a/tests/test_atomic_limits.py +++ b/tests/test_atomic_limits.py @@ -48,3 +48,46 @@ def test_google_exhaustion_never_calls_verifier(client, monkeypatch): assert response.status_code == 429 assert int(response.headers['retry-after']) >= 1 verify.assert_not_called() + + +def test_concurrent_uploads_cannot_exceed_storage_quota(): + from backend.services.resume_files import ResumeFileManager + from backend.services.object_storage import ObjectStorageService + + user = main.storage.create_user("quota_race@example.test", None, "Quota Race", email_verified=True) + obj_storage = ObjectStorageService(endpoint_url=None) + manager = ResumeFileManager(main.storage, obj_storage) + + max_quota = 100 + file_bytes = b"A" * 70 + start = Barrier(2) + + def upload_task(idx): + start.wait() + try: + resume = manager.reserve_and_upload( + user_id=user.id, + filename=f"resume_{idx}.txt", + content_type="text/plain", + content_bytes=file_bytes, + resume_name=f"Resume {idx}", + text_content=f"Content {idx}", + max_storage_bytes=max_quota, + ) + return ("success", resume) + except Exception as e: + return ("error", str(e)) + + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(upload_task, range(2))) + + successes = [r for r in results if r[0] == "success"] + errors = [r for r in results if r[0] == "error"] + + assert len(successes) == 1, f"Expected exactly 1 success, got {len(successes)}: {results}" + assert len(errors) == 1, f"Expected exactly 1 error, got {len(errors)}: {results}" + assert "storage limit" in errors[0][1].lower() + + total_bytes = main.storage.get_user_upload_bytes(user.id) + assert total_bytes == 70 + assert total_bytes <= max_quota From 44d4c5105bf999524360c2f3f44b4fb0e879425f Mon Sep 17 00:00:00 2001 From: Sourav Chandhok <114119117+souravC01@users.noreply.github.com> Date: Sun, 13 Sep 2026 23:00:42 -0400 Subject: [PATCH 4/4] fix(auth): consume recovery tokens and update passwords atomically Validate, consume (used_at IS NULL), and apply password changes in a single database transaction, preventing concurrent recovery requests from overwriting passwords multiple times. Add single-transaction email verification consumption and automated concurrency test coverage. --- backend/routers/auth.py | 27 ++-------- backend/storage.py | 101 +++++++++++++++++++++++++++++++++++- tests/test_atomic_limits.py | 31 +++++++++++ 3 files changed, 135 insertions(+), 24 deletions(-) diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 679043b..03d20ab 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -355,20 +355,11 @@ def confirm_email_verification(req: EmailVerificationConfirm): clean_token = req.token.strip() token_hash = hashlib.sha256(clean_token.encode("utf-8")).hexdigest() storage = get_storage() - record = storage.get_auth_token(token_hash, "verify_email") + user_id = storage.consume_verify_token_and_verify_user(token_hash) - if not record: + if not user_id: raise HTTPException(status_code=400, detail="Invalid or expired verification link.") - now_iso = datetime.now(timezone.utc).isoformat() - if record["expires_at"] < now_iso: - raise HTTPException(status_code=400, detail="Verification link has expired. Please request a new one.") - - storage.consume_auth_token(record["id"]) - storage.invalidate_prior_auth_tokens(record["user_id"], "verify_email") - storage.update_user_verification(record["user_id"], verified=True) - storage.increment_user_session_version(record["user_id"]) - return {"success": True, "message": "Email verified successfully. You may now log in."} @@ -425,20 +416,12 @@ def confirm_password_reset(req: PasswordResetConfirm): clean_token = req.token.strip() token_hash = hashlib.sha256(clean_token.encode("utf-8")).hexdigest() storage = get_storage() - record = storage.get_auth_token(token_hash, "reset_password") + hashed = hash_password(req.new_password) + user_id = storage.consume_reset_token_and_update_password(token_hash, hashed) - if not record: + if not user_id: raise HTTPException(status_code=400, detail="Invalid or expired password reset link.") - now_iso = datetime.now(timezone.utc).isoformat() - if record["expires_at"] < now_iso: - raise HTTPException(status_code=400, detail="Password reset link has expired. Please request a new one.") - - hashed = hash_password(req.new_password) - storage.consume_auth_token(record["id"]) - storage.invalidate_prior_auth_tokens(record["user_id"], "reset_password") - storage.update_user_password(record["user_id"], hashed, increment_session=True) - return {"success": True, "message": "Password updated successfully. Please log in with your new password."} diff --git a/backend/storage.py b/backend/storage.py index cff790c..9461bf7 100644 --- a/backend/storage.py +++ b/backend/storage.py @@ -542,13 +542,110 @@ def get_auth_token(self, token_hash: str, token_type: str) -> Optional[Dict[str, return None return dict(row) - def consume_auth_token(self, token_id: str) -> None: + def consume_auth_token(self, token_id: str) -> bool: now = datetime.now().isoformat() with self._get_cursor() as cursor: cursor.execute( - self._format_sql("UPDATE auth_tokens SET used_at = ? WHERE id = ?"), + self._format_sql("UPDATE auth_tokens SET used_at = ? WHERE id = ? AND used_at IS NULL"), (now, token_id), ) + return cursor.rowcount > 0 + + def consume_reset_token_and_update_password(self, token_hash: str, new_hashed_password: str) -> Optional[str]: + now = datetime.now().isoformat() + now_utc_iso = datetime.now(timezone.utc).isoformat() + with self._get_immediate_cursor() as cursor: + if self.is_postgres: + cursor.execute( + "SELECT id, user_id, expires_at, used_at FROM auth_tokens WHERE token_hash = %s AND token_type = 'reset_password' FOR UPDATE", + (token_hash,), + ) + else: + cursor.execute( + self._format_sql( + "SELECT id, user_id, expires_at, used_at FROM auth_tokens WHERE token_hash = ? AND token_type = 'reset_password'" + ), + (token_hash,), + ) + row = cursor.fetchone() + if not row: + return None + token_id = row["id"] if isinstance(row, dict) else row[0] + user_id = row["user_id"] if isinstance(row, dict) else row[1] + expires_at = row["expires_at"] if isinstance(row, dict) else row[2] + used_at = row["used_at"] if isinstance(row, dict) else row[3] + + if used_at is not None or expires_at < now_utc_iso: + return None + + cursor.execute( + self._format_sql("UPDATE auth_tokens SET used_at = ? WHERE id = ? AND used_at IS NULL"), + (now, token_id), + ) + if cursor.rowcount != 1: + return None + + cursor.execute( + self._format_sql( + "UPDATE auth_tokens SET used_at = ? WHERE user_id = ? AND token_type = 'reset_password' AND used_at IS NULL" + ), + (now, user_id), + ) + cursor.execute( + self._format_sql( + "UPDATE users SET hashed_password = ?, session_version = session_version + 1, updated_at = ? WHERE id = ?" + ), + (new_hashed_password, now, user_id), + ) + return user_id + + def consume_verify_token_and_verify_user(self, token_hash: str) -> Optional[str]: + now = datetime.now().isoformat() + now_utc_iso = datetime.now(timezone.utc).isoformat() + with self._get_immediate_cursor() as cursor: + if self.is_postgres: + cursor.execute( + "SELECT id, user_id, expires_at, used_at FROM auth_tokens WHERE token_hash = %s AND token_type = 'verify_email' FOR UPDATE", + (token_hash,), + ) + else: + cursor.execute( + self._format_sql( + "SELECT id, user_id, expires_at, used_at FROM auth_tokens WHERE token_hash = ? AND token_type = 'verify_email'" + ), + (token_hash,), + ) + row = cursor.fetchone() + if not row: + return None + token_id = row["id"] if isinstance(row, dict) else row[0] + user_id = row["user_id"] if isinstance(row, dict) else row[1] + expires_at = row["expires_at"] if isinstance(row, dict) else row[2] + used_at = row["used_at"] if isinstance(row, dict) else row[3] + + if used_at is not None or expires_at < now_utc_iso: + return None + + cursor.execute( + self._format_sql("UPDATE auth_tokens SET used_at = ? WHERE id = ? AND used_at IS NULL"), + (now, token_id), + ) + if cursor.rowcount != 1: + return None + + cursor.execute( + self._format_sql( + "UPDATE auth_tokens SET used_at = ? WHERE user_id = ? AND token_type = 'verify_email' AND used_at IS NULL" + ), + (now, user_id), + ) + cursor.execute( + self._format_sql( + "UPDATE users SET email_verified = TRUE, session_version = session_version + 1, updated_at = ? WHERE id = ?" + ), + (now, user_id), + ) + return user_id def invalidate_prior_auth_tokens(self, user_id: str, token_type: str) -> None: now = datetime.now().isoformat() diff --git a/tests/test_atomic_limits.py b/tests/test_atomic_limits.py index 23b7359..e8f5272 100644 --- a/tests/test_atomic_limits.py +++ b/tests/test_atomic_limits.py @@ -91,3 +91,34 @@ def upload_task(idx): total_bytes = main.storage.get_user_upload_bytes(user.id) assert total_bytes == 70 assert total_bytes <= max_quota + + +def test_concurrent_password_reset_confirm_only_succeeds_once(client): + import secrets + import hashlib + from datetime import datetime, timezone, timedelta + + user = main.storage.create_user("reset_race@example.test", "OldPassword1!", "Reset Race", email_verified=True) + raw_token = secrets.token_urlsafe(32) + token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest() + expires_at = (datetime.now(timezone.utc) + timedelta(minutes=30)).isoformat() + main.storage.create_auth_token(user.id, token_hash, "reset_password", expires_at) + + start = Barrier(2) + + def confirm_task(idx): + start.wait() + res = client.post( + "/api/auth/password-reset/confirm", + json={"token": raw_token, "new_password": f"NewPassword{idx}!"}, + ) + return res.status_code, res.json() + + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(confirm_task, range(2))) + + status_codes = [r[0] for r in results] + assert status_codes.count(200) == 1, f"Expected exactly one 200, got {results}" + assert status_codes.count(400) == 1, f"Expected exactly one 400, got {results}" + error_res = next(r[1] for r in results if r[0] == 400) + assert "invalid or expired" in error_res["detail"].lower()