CWE-362: Concurrent Execution using Shared Resource with Improper Synchronization ('Race Condition') - Python

Overview

The GIL prevents two threads from executing Python bytecode at the exact same instant, but it does not make compound operations atomic: counter += 1, a dict/list mutation split across multiple statements, or a check-then-act sequence can still be interrupted between bytecode instructions and interleaved with another thread. The primary fix for threads is threading.Lock/threading.RLock around the full read-modify-write sequence. asyncio code has a different but equally real hazard: coroutines never run truly in parallel, but an await on something that actually suspends is a point where another task can run and mutate shared state before the current coroutine resumes, so an unguarded read-await-write across a scheduling point is a race even in single-threaded asyncio - use asyncio.Lock there. For multiple OS processes (multiprocessing, separate workers, or separate application server instances), the GIL does not apply at all, and shared state needs multiprocessing.Lock/Value, or - the common case for a web application - should be pushed to the database with SELECT ... FOR UPDATE or an atomic conditional UPDATE.

Common Vulnerable Patterns

Compound Operation Assumed Atomic Because of the GIL

import threading

# VULNERABLE - counter += 1 is not atomic, despite the GIL
class RequestCounter:
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1  # read, add, write: three separate bytecode steps

# Attack: 1,000 threads each call increment() once.
# Result: counter.count often ends up less than 1,000 - the GIL can switch
# threads between the read and the write of `count`.

Why this is vulnerable: count += 1 compiles to a LOAD_FAST, a binary add, and a STORE_FAST - the GIL can release control between these bytecode instructions (it switches periodically, and I/O or certain built-in calls can also trigger a switch), so two threads can both read the same starting value before either writes back the incremented result.

Coroutine Mutating State Across an await

import asyncio

# VULNERABLE - check and update are separated by an await
class AsyncAccount:
    def __init__(self, balance):
        self.balance = balance

    async def withdraw(self, amount):
        if self.balance < amount:
            raise ValueError("insufficient funds")
        await self._log_transaction(amount)  # RACE WINDOW: suspension point
        self.balance -= amount

    async def _log_transaction(self, amount):
        # A real await point: a database round trip, an HTTP call, a queue put.
        # asyncio.sleep(0) is the smallest thing that yields to the event loop.
        await asyncio.sleep(0)

# Attack: two concurrent await account.withdraw(100) calls when balance = 100.
# Both pass the check before either awaited call resumes and decrements the
# balance. Result: balance ends at -100.

Why this is vulnerable: asyncio runs one coroutine at a time, but await yields control back to the event loop, and another task - including a second call to withdraw - can run to completion during that suspension. The check made before the await is stale by the time execution resumes. Measured on Python 3.13, asyncio.gather of five withdraw(30) calls against a balance of 100 returned five successes on 100 runs out of 100, ending at -50.

await is not itself the suspension point - the awaited object is. Awaiting a coroutine that never awaits anything does not reach the event loop at all: the call runs to completion inline, exactly like a synchronous one, and this race cannot occur. Measured on Python 3.13 with _log_transaction's body written as ..., a concurrently running ticker task advanced 0 times across 1,000 await calls, and the same gather of five withdrawals correctly rejected two of them. That matters for triage. A SELECT through an async driver, an aiohttp request, an asyncio.Queue.put on a full queue and asyncio.sleep(0) all yield; an async def that only does arithmetic, and an await on an already-resolved Future, do not. Before reporting a read-await-write as exploitable, check that the awaited call has a real yield point in it. Before concluding it is safe, weigh how easily one arrives later: adding a yield point is a one-line change nobody reviews as a concurrency change.

Separate SELECT Then UPDATE Without a Transaction

# VULNERABLE - the read and the write are two independent database round-trips
def withdraw(conn, account_id, amount):
    cursor = conn.cursor()
    cursor.execute("SELECT balance FROM accounts WHERE id = %s", (account_id,))
    balance = cursor.fetchone()[0]

    if balance < amount:
        raise ValueError("insufficient funds")

    # RACE WINDOW: another request's SELECT/UPDATE pair can run here
    cursor.execute(
        "UPDATE accounts SET balance = balance - %s WHERE id = %s",
        (amount, account_id),
    )
    conn.commit()

# Attack: two concurrent withdraw(conn, account_id, 100) calls when the
# balance is 100. Both SELECTs see balance = 100, both UPDATEs subtract 100.
# Result: the balance ends at -100 instead of the second call being rejected.

Why this is vulnerable: The SELECT and UPDATE run as two separate statements with no lock or transaction tying them together, so a concurrent request's SELECT can read the same pre-deduction balance before this request's UPDATE commits.

Secure Patterns

threading.Lock for a Multi-Threaded Critical Section

import threading

# SECURE - threading.Lock protects the full read-modify-write sequence
class Counter:
    def __init__(self):
        self._lock = threading.Lock()
        self._value = 0

    def increment(self):
        with self._lock:
            self._value += 1

    @property
    def value(self):
        with self._lock:
            return self._value

Why this works: with self._lock: acquires the lock before the block and releases it on exit, even if an exception is raised inside - only one thread can execute the block at a time, so the read, add, and write always happen as one atomic unit relative to other threads. Reading _value through the same lock, not just writing it, matters too: an unguarded read can observe a partially-updated value on interpreter implementations or object types where the GIL alone would not guarantee visibility.

asyncio.Lock for an Async Critical Section

import asyncio

# SECURE - asyncio.Lock prevents another task from interleaving across an await point
class AsyncAccount:
    def __init__(self, balance):
        self._lock = asyncio.Lock()
        self._balance = balance

    @property
    def balance(self):
        # exposed so a test can assert on the outcome, not only on how many
        # callers believed they succeeded
        return self._balance

    async def withdraw(self, amount):
        async with self._lock:
            if self._balance < amount:
                raise ValueError("insufficient funds")
            await self._log_transaction(amount)
            self._balance -= amount

    async def _log_transaction(self, amount):
        # a real await point, the same one the vulnerable example has
        await asyncio.sleep(0)

Why this works: async with self._lock: suspends any other coroutine trying to enter the block until the current holder exits, including across the await self._log_transaction(amount) inside it - a second concurrent call to withdraw waits for the lock rather than interleaving with the first. This closes the race window a plain await between the check and the write would otherwise leave open. Measured on Python 3.13: asyncio.gather of five withdraw(30) calls against a balance of 100 returned exactly three successes and a final balance of 10, on 100 runs out of 100 - against five successes and -50 for the unlocked version.

async with also releases on the exception path, so the raise ValueError inside the block does not leave the lock held. That is worth asserting rather than assuming: a hand-rolled await lock.acquire() followed by lock.release() without a try/finally deadlocks every later caller the first time a withdrawal is refused, and every rejection test still passes.

One asyncio.Lock per resource, created where the resource lives. The lock here is an attribute of the account it protects. A lock created inside the request handler, or per AsyncAccount instance where a new instance is built per request from the same underlying row, is a different object for every caller and excludes nobody - the code reads as locked and behaves as if it were not.

multiprocessing.Lock and Value for Cross-Process State

from multiprocessing import Process, Lock, Value

# SECURE - multiprocessing.Value with a lock protects a counter shared
# across OS processes, where threading.Lock cannot reach
def worker(counter, lock):
    for _ in range(1000):
        with lock:
            counter.value += 1

if __name__ == "__main__":
    lock = Lock()
    counter = Value("i", 0)  # shared integer, synchronized manually via lock
    processes = [Process(target=worker, args=(counter, lock)) for _ in range(4)]

    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print(counter.value)  # always 4000

Why this works: threading.Lock only coordinates threads inside one process; it does nothing for separate OS processes, which do not share Python-level memory at all. multiprocessing.Value allocates the integer in shared memory that every process can see, and the explicit Lock (Value's own built-in lock is also available via counter.get_lock()) serializes the read-modify-write across all of them.

Database Transaction with SELECT ... FOR UPDATE

# SECURE - the row lock is held from the SELECT through the UPDATE, inside
# one transaction, so no other transaction can read a stale balance
def withdraw(conn, account_id, amount):
    with conn:  # commits on success, rolls back on exception
        cursor = conn.cursor()
        cursor.execute(
            "SELECT balance FROM accounts WHERE id = %s FOR UPDATE", (account_id,)
        )
        row = cursor.fetchone()
        if row is None:
            raise LookupError("account not found")

        balance = row[0]
        if balance < amount:
            raise ValueError("insufficient funds")

        cursor.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, account_id),
        )

Why this works: FOR UPDATE takes a row-level lock at the SELECT, and the transaction holds that lock until commit or rollback. Any other transaction's SELECT ... FOR UPDATE on the same row blocks until this one finishes, so the balance this code checked cannot change out from under it before the UPDATE runs - correct across every process and server instance connected to the database, not just within one Python process.

Conditional Atomic UPDATE (No Row Lock Needed)

# SECURE - the precondition and the write happen in one atomic statement
def withdraw_atomic(conn, account_id, amount):
    with conn:
        cursor = conn.cursor()
        cursor.execute(
            """
            UPDATE accounts SET balance = balance - %s
            WHERE id = %s AND balance >= %s
            """,
            (amount, account_id, amount),
        )
        if cursor.rowcount == 0:
            raise ValueError("insufficient funds or account not found")

Why this works: There is no separate read at all - the database evaluates balance >= %s and performs the subtraction as part of the same atomic statement. cursor.rowcount == 0 separates "the precondition failed" from "the write succeeded" here, without needing an explicit lock for this simple case. On MySQL drivers this counts rows changed rather than matched unless the connection requests FOUND_ROWS, so an UPDATE writing a value a row already holds reports 0; a non-zero withdrawal always changes the balance, so this pattern is unaffected.

Django ORM: select_for_update and F() Expressions

from django.db import transaction
from django.db.models import F

# SECURE - select_for_update() locks the row for the transaction's duration
def withdraw(account_id, amount):
    with transaction.atomic():
        account = Account.objects.select_for_update().get(id=account_id)
        if account.balance < amount:
            raise ValueError("insufficient funds")
        account.balance -= amount
        account.save()

# SECURE - F() pushes the arithmetic into the database as a single atomic
# UPDATE, avoiding the need for an explicit lock entirely
def withdraw_atomic(account_id, amount):
    updated = Account.objects.filter(
        id=account_id, balance__gte=amount
    ).update(balance=F("balance") - amount)

    if updated == 0:
        raise ValueError("insufficient funds or account not found")

Why this works: select_for_update() inside transaction.atomic() issues SELECT ... FOR UPDATE, holding a row lock for the duration of the transaction just like the raw SQL pattern above. The F() expression version goes further: balance=F("balance") - amount tells the database to compute the new value from its own current value as part of the UPDATE statement, so Django never reads the balance into Python at all - there is no window for a stale read to exist in.

Framework-Specific Guidance

Django

# SECURE - select_for_update(nowait=True) fails fast instead of blocking when
# a row is already locked, useful for user-facing requests that should not hang
from django.db import transaction, DatabaseError

def try_withdraw(account_id, amount):
    try:
        with transaction.atomic():
            account = Account.objects.select_for_update(nowait=True).get(id=account_id)
            if account.balance < amount:
                return False
            account.balance -= amount
            account.save()
            return True
    except DatabaseError:
        return False  # row is locked by another transaction right now

Why this works: nowait=True raises immediately instead of waiting for a contended lock to release, which is often preferable for a user-facing request path over letting it block indefinitely; the caller can surface a "please retry" response rather than tying up a request thread.

Flask / FastAPI with SQLAlchemy

# SECURE - SQLAlchemy's with_for_update() for pessimistic locking
from sqlalchemy import select

def withdraw(session, account_id: int, amount: int) -> None:
    with session.begin():
        account = session.execute(
            select(Account).where(Account.id == account_id).with_for_update()
        ).scalar_one()

        if account.balance < amount:
            raise ValueError("insufficient funds")

        account.balance -= amount

Why this works: with_for_update() adds FOR UPDATE to the generated query, and session.begin() wraps the read and write in one transaction, so the row lock is held for the whole critical section regardless of how many application server processes or async workers are handling requests concurrently.

Considerations

A lock is only as wide as the thing holding it. An in-process lock serialises the threads inside one instance and does nothing about a second instance, so a fix that works on a developer machine can fail the moment the service is scaled out or restarted behind a load balancer. Decide first where the shared state actually lives. If it is a database row, the serialisation has to happen in the database - a row lock, a conditional update, or a version column. If it is genuinely in-process and stays that way, an in-process lock is correct and cheaper.

Not every race is worth fixing. Two requests overwriting a display preference, or a page-view counter losing an increment, is a race with no security consequence and often no user-visible one. The ones that matter change a decision: a balance check, a quota, a one-time token being redeemed, a permission being evaluated. Fixing a benign race costs throughput and adds a failure mode, so say which category the finding is in before reaching for a lock.

The GIL does not make compound operations atomic. It guarantees that a single bytecode does not tear, not that a read-modify-write is indivisible; counter += 1 is several bytecodes and interleaves freely. Match the lock to the concurrency model as well: threading.Lock does nothing across processes, asyncio.Lock does nothing across threads, and a Gunicorn deployment with multiple workers has both.

Optimistic and pessimistic locking fail in opposite directions. A pessimistic lock makes every caller wait, so it is predictable but caps throughput and can deadlock if two paths take locks in different orders. Optimistic concurrency lets callers proceed and rejects the loser, which is faster when conflicts are rare and degenerates into wasted work and retries when they are common. Pick by how often the same row is genuinely contended, not by which is easier to write.

Retries need a bound and a backoff. A conflict-and-retry loop with neither turns a contended row into a livelock under load - every caller retrying immediately, none making progress. Cap the attempts, back off between them, and decide what the caller sees when the cap is reached. "Try again" is a legitimate answer; silently returning stale data is not.

Testing

A re-scan cannot confirm this fix. The tool sees a threading.Lock where there was none and reports the finding closed; it cannot tell whether the lock covers the whole decision, whether every caller takes the same lock object, or whether the locked version still serves a legitimate request.

The two concurrency models need different tests, and only one of them is deterministic out of the box.

asyncio: gather is enough, and reliably so

Every await on something that actually suspends is a guaranteed yield to the event loop, so an asyncio.gather of concurrent callers reproduces the race on every run rather than occasionally. Measured on Python 3.13 over 100 runs each: the unlocked coroutine returned 5 successes every time, the asyncio.Lock version 3 every time.

import asyncio
import pytest
# The lock-guarded account from the section above, as its own module
from account import AsyncAccount

@pytest.mark.asyncio   # requires pytest-asyncio; anyio's @pytest.mark.anyio also works
async def test_concurrent_async_withdrawals_never_overdraft():
    account = AsyncAccount(100)

    async def attempt():
        try:
            await account.withdraw(30)
            return True
        except ValueError:
            return False

    results = await asyncio.gather(*(attempt() for _ in range(5)))

    # Exactly 3 of 5 concurrent withdrawals of 30 succeed against 100,
    # and the balance is the arithmetic those 3 imply.
    assert sum(results) == 3
    assert account.balance == 10

Check that the test actually ran, because what happens without an async plugin depends on the pytest version. Measured with a deliberately failing async def test and no plugin: pytest 8.3.5 emits PytestUnhandledCoroutineWarning, skips it, and exits 0 - a green run for a test that asserts 1 == 2. pytest 9.1.1 fails it outright with the same message. On the older behaviour a whole concurrency suite can be silently inert, so break one assertion once and confirm it goes red.

threads: launching them together proves nothing

The same test written with threading.Thread does not reproduce the race at all. The GIL switches threads every 5 ms by default, and an unsynchronized withdraw completes in far less than that, so each thread runs the whole check-then-act before the next one starts. Measured on Python 3.13: five threads against the unlocked account returned the "correct" answer of 3 on 500 runs out of 500. A test that never fails against the bug is not a test.

Pin the interleaving. Give the class under test a hook between the check and the write, and make the hook a threading.Barrier: every thread that passed the check waits there until all of them have, so the stale read is guaranteed. Give the barrier a timeout so a correctly locked implementation - where only one thread is ever inside the section - waits it out and completes rather than deadlocking.

import threading

class Account:
    """Production code passes no hook; only the test does."""

    def __init__(self, balance, between_check_and_write=None):
        self._lock = threading.Lock()
        self.balance = balance
        self._between_check_and_write = between_check_and_write

    def withdraw(self, amount):
        with self._lock:
            if self.balance < amount:
                raise ValueError("insufficient funds")
            if self._between_check_and_write:
                self._between_check_and_write()
            self.balance -= amount


def test_concurrent_withdrawals_never_overdraft():
    barrier = threading.Barrier(5, timeout=0.5)

    def between_check_and_write():
        try:
            barrier.wait()
        except threading.BrokenBarrierError:
            pass  # only one thread got here: the lock is doing its job

    account = Account(100, between_check_and_write)
    results = []

    def attempt():
        try:
            account.withdraw(30)
            results.append(True)
        except ValueError:
            results.append(False)

    threads = [threading.Thread(target=attempt) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert sum(results) == 3
    assert account.balance == 10

Measured on Python 3.13 over 50 runs, this returns 5 successes every time against an Account without the with self._lock, and 3 every time with it.

Assert these, with the result each should produce:

• Accept, single call. withdraw(30) against 100 returns and leaves balance at exactly 70. A lock that refuses every caller passes the concurrency test above and fails this one.
• Accept, at the boundary. withdraw(100) against 100 succeeds and leaves 0.
• Reject, past the boundary. withdraw(1) against 0 raises ValueError and leaves balance at 0 - unchanged, not merely non-negative.
• Concurrent. The barrier test above: sum(results) == 3 and balance == 10. Assert the balance as well as the count; a lost update can produce the right number of successes and the wrong total.
• The lock is released on the rejection path. After a refused withdrawal, a later legitimate withdraw on the same object still returns. A lock.acquire()/lock.release() pair without try/finally deadlocks here and passes every other assertion.
• The lock is the same object for every caller. Two callers that reach the same resource through different code paths contend for one lock. Where the object under test is rebuilt per request, this is where the fix fails, and the barrier test above will not catch it because the test constructs one instance.

For cross-process state, none of the above applies: threading.Lock and asyncio.Lock are invisible across processes. Drive the real endpoint from several processes at once and assert on the persisted state.

Common Pitfalls

• Trusting the GIL to make a compound operation atomic: counter += 1, dict[key] = dict.get(key, 0) + 1, and similar patterns are not atomic just because only one thread runs Python bytecode at a time - the GIL can switch between the read and the write. Wrap the operation in threading.Lock regardless.
• Using threading.Lock inside asyncio code: A standard threading.Lock's blocking acquire() stalls the entire event loop, not just the current coroutine, defeating the purpose of async/await; use asyncio.Lock for coroutine-based code instead.
• Guarding the check with a lock, then awaiting or performing I/O outside it before writing: async with lock: check(); followed later by an unguarded self._balance -= amount still leaves the gap open; the entire check-then-write sequence, including any await inside it, needs to be inside the same async with lock: block.
• Assuming multiprocessing.Value's built-in lock covers a compound operation across multiple calls: counter.value += 1 on a Value is not automatically locked just because the object supports locking - the increment still needs an explicit with counter.get_lock(): (or the pattern shown above) around the read-modify-write.

Dependencies and Installation

No additional package is required for threading.Lock, asyncio.Lock, or multiprocessing.Lock/Value - all are part of the standard library. select_for_update() requires Django's ORM (already a dependency for a Django project); with_for_update() requires SQLAlchemy (already a dependency for a Flask/FastAPI project using it as the ORM).

The async test in the Testing section needs a pytest plugin, because pytest cannot run an async def test on its own:

pip install pytest-asyncio     # or: pip install anyio

Additional Resources

• CWE-362 Details
• Python threading documentation
• Python asyncio synchronization primitives
• Python multiprocessing documentation
• Django documentation: select_for_update()