Fix NoSQL operator injection in Python and Flask

1. Find every PyMongo call whose filter or update takes request data: find, find_one, find_one_and_update, update_one, delete_many, count_documents and aggregate with request.get_json(), request.args, request.form or headers, including in the data-access modules your views call.
2. Check the shape of the input before the query: each value must be a str (or an int) of a sane length; answer 400 for dict, list, None and bool. A schema such as Pydantic with StrictStr does the same job.
3. Write filters as {"email": {"$eq": email}}, so that even a value that slipped past validation is compared literally.
4. Look an account up by its identifier and compare secrets (reset tokens, invitation codes, API keys) in your code with hmac.compare_digest, then consume the token atomically with update_one filtered on _id and the token.
5. Never pass a dict from the request as a whole filter, projection, sort or update; build them from an allowlist of fields.
6. If nothing uses $where, $function or $accumulator, turn server-side JavaScript off on the server (security.javascriptEnabled: false).
7. Add regression tests that send operator payloads ($ne, $gt, $gte, $regex, $exists, $nin, $type, $in) and check that nothing changes.

The three implementations share one API and one test suite; only the code below changes. The tests load the one that IMPL names.

def reset_password(users, body: dict, new_password_hash: str, now) -> bool:
    user = users.find_one_and_update(
        {"email": body.get("email"), "resetToken": body.get("token"), "resetExpires": {"$gt": now}},
        {"$set": {"passwordHash": new_password_hash}, "$unset": {"resetToken": "", "resetExpires": ""}},
    )
    return user is not None

def reset_password(users, body: dict, new_password_hash: str, now) -> bool:
    email, token = body.get("email"), body.get("token")
    # 1. A JSON value can be a dict, a list, a number or None: accept strings only.
    if not isinstance(email, str) or not isinstance(token, str):
        return False
    if len(email) > 254 or len(token) > 128:
        return False

    # 2. Look the account up by email alone, with $eq, so the value is always a literal.
    user = users.find_one({"email": {"$eq": email}})

    # 3. Check expiry and the token in code; the database never evaluates the caller's token.
    if user is None or not isinstance(user.get("resetToken"), str):
        return False
    if user.get("resetExpires") is None or user["resetExpires"] <= now:
        return False
    if not hmac.compare_digest(token.encode(), user["resetToken"].encode()):
        return False

    # 4. Consume the token atomically: a replay or a parallel request updates nothing.
    result = users.update_one(
        {"_id": user["_id"], "resetToken": {"$eq": token}},
        {"$set": {"passwordHash": new_password_hash}, "$unset": {"resetToken": "", "resetExpires": ""}},
    )
    return result.modified_count == 1

BLOCKED_OPERATORS = {"$ne", "$gt", "$regex", "$where"}


def contains_blocked_operator(value) -> bool:
    return isinstance(value, dict) and any(key in BLOCKED_OPERATORS for key in value)


def reset_password(users, body: dict, new_password_hash: str, now) -> bool:
    email, token = body.get("email"), body.get("token")
    if contains_blocked_operator(email) or contains_blocked_operator(token):
        return False
    user = users.find_one_and_update(
        {"email": email, "resetToken": token, "resetExpires": {"$gt": now}},
        {"$set": {"passwordHash": new_password_hash}, "$unset": {"resetToken": "", "resetExpires": ""}},
    )
    return user is not None

The exploit tests send the attack and check that it fails; functional tests next to them keep ordinary requests working, so a fix that rejects everything fails too.

class TestExploit:
    """The attacker knows Alice's email address but has never seen the reset email. Each payload
    must be refused, and no account may change."""

    @pytest.fixture()
    def assert_refused(self, post, sign_in, snapshot, alice):
        def check(body: dict):
            before = snapshot()
            status, response = post("/reset-password", {"password": ATTACKER_PASSWORD, **body})
            assert status == 400, f"reset accepted: {response}"
            assert snapshot() == before, "an account changed"
            assert sign_in(alice["email"], ATTACKER_PASSWORD) == 401

        return check

    def test_ne_null_as_the_token_does_not_mean_any_token(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$ne": None}})

    def test_gt_empty_string_as_the_token_does_not_match_every_string(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$gt": ""}})

    def test_regex_cannot_guess_the_token_one_character_at_a_time(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$regex": "^K"}})

    def test_exists_true_as_the_token_is_refused(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$exists": True}})

    def test_gte_empty_string_as_the_token_is_refused(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$gte": ""}})

    def test_nin_empty_list_as_the_token_is_refused(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$nin": []}})

    def test_type_string_as_the_token_is_refused(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$type": "string"}})

    def test_in_cannot_try_many_guesses_in_one_request(self, assert_refused, alice):
        assert_refused({"email": alice["email"], "token": {"$in": ["guess-1", "guess-2", alice["resetToken"]]}})

    def test_operators_in_both_fields_cannot_reset_an_unknown_account(self, assert_refused):
        assert_refused({"email": {"$gte": ""}, "token": {"$exists": True}})

• Blocklists of the operators in the write-ups ($ne, $gt, $regex, $where) miss $gte, $exists, $nin, $type and $in, which match just as well: this card's wrong fix fails six exploit tests that way.
• request.get_json() returns whatever JSON the client sent, and body.get("token") is a dict, list, int or None as often as an attacker likes. Type hints on your function do not check anything at run time.
• request.args and request.form values are always strings in Flask, but code that runs json.loads() on a query parameter, or a framework that parses token[$ne]=x into nested objects, brings the problem back.
• str(value) turns a dict into "{'$ne': None}", which stops the operator but hides the bug; validate the type and answer 400 instead.
• $where, $function and $accumulator run JavaScript on the database server, and MongoDB enables server-side scripting by default. They are reachable when the request controls a whole filter, as in collection.find(request.json["filter"]).
• mongomock is a test double, not MongoDB: it implements the operators used here, and its own Missing Features list names $where, $text and $jsonSchema among the query operators it lacks. This card's payloads and queries were run against both mongomock 4.3.0 and mongod 8.0.32 through PyMongo 4.18.2 and gave the same matches and the same final documents.
• OWASP's Forgot Password Cheat Sheet asks for reset tokens that are random, long, stored securely, single use and short-lived. This card stores the token as the vulnerable app does; storing a SHA-256 hash of it instead keeps a database leak from handing out working reset links.

1. Run the tests in all three modes: IMPL=fixed must pass; IMPL=vulnerable and IMPL=wrong must fail (node ../../run-all.mjs nosql-injection/python-flask does this for you).
2. Against your own endpoint and a real MongoDB, post {"email": "<a test account>", "token": {"$ne": null}, "password": "..."}, then the same with {"$gte": ""} and {"$exists": true}, and expect 400 with the account unchanged.
3. Static analysis is not a control here: Semgrep's community packs and Bandit have no Python MongoDB injection rule, and CodeQL's py/nosql-injection models PyMongo but not mongomock and was not run. The regression tests are the evidence.

What each rule reported when it was run on this card's code, or what its source shows it covers where it was not run. A rule that exists for the weakness does not always catch it.

Download the project, install its dependencies once (this step needs the network), then run the tests against each implementation from the nosql-injection/python-flaskfolder.

python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt

# Expected: every test passes
IMPL=fixed .venv/bin/python -m pytest

# Expected: an exploit test fails
IMPL=vulnerable .venv/bin/python -m pytest

# Expected: a test fails
IMPL=wrong .venv/bin/python -m pytest

Each run writes a JUnit report to reports/junit.xml. From the top folder of the download, node run-all.mjs nosql-injection/python-flask runs all three modes and checks the expected results.

Download the project ZIP, 20 KB

Verified

Machine

macOS 27.0 (Darwin 27.0.0, arm64)

Toolchain

• python CPython 3.14.8
• flask 3.1.3
• pymongo 4.18.2
• mongomock 4.3.0
• pytest 9.1.1
• parityCheck mongod 8.0.32 (arm64 macOS build) through PyMongo 4.18.2
• semgrep 1.180.0
• bandit 1.9.4

Matrix

fixed: 16 of 16 tests pass; vulnerable: 9 of 16 tests fail (9 exploit tests); wrong: 6 of 16 tests fail (6 exploit tests)

Test results

fixed 16/16 pass; vulnerable 9/16 fail (9 exploit tests); wrong 6/16 fail (6 exploit tests: $exists, $gte, $nin, $type, $in and operators in both fields), offline after install

Dependency freshness

pymongo 4.18.2 (2026-09-24) fixes CVE-2026-96747, CVE-2026-96748 and CVE-2026-96749 (advisories published 2026-10-05); 4.18.3 (2026-10-08) is a bug-fix release two days old, so the card stays on 4.18.2. dnspython is pinned to 2.8.0 and pytz to 2026.4 because 2.9.0 and 2026.5 were under seven days old. mongomock 4.3.0 (2024-11-16) is the latest release; the project is quiet but installs and runs cleanly on Python 3.14.8 with PyMongo 4.18.2. Rerun on CPython 3.14.8, the 30 September 2026 expedited security release.

Versions and results hold for the toolchain above. The card's tests are the check: run them with your own versions before relying on the fix.

References

1. OWASP NoSQL Security Cheat Sheet
2. OWASP WSTG: Testing for NoSQL Injection
3. MongoDB manual: $eq (implicit and explicit equality)
4. MongoDB manual: server-side JavaScript (security.javascriptEnabled, --noscripting)
5. PyMongo: Collection API (find_one_and_update, update_one)
6. PyMongo changelog: 4.18.2 security fixes
7. Python: hmac.compare_digest()
8. Flask: Request.get_json()
9. mongomock: Missing Features
10. OWASP Forgot Password Cheat Sheet
11. CWE-943: Improper Neutralization of Special Elements in Data Query Logic
12. CAPEC-676: NoSQL Injection

Attack patterns:

• CAPEC-676 at MITRE