From 770b7792b456a29a1e23bd07754ed610c6f5b7ab Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:09:48 -0700 Subject: [PATCH 01/11] Fix BOLA: reject password change for another user's account PUT /users/v1/{username}/password previously used the URL path username directly to look up and overwrite a user's password, allowing any authenticated user to hijack any other account by supplying a different username in the URL while presenting their own valid JWT. Now the handler compares the URL username against the authenticated caller's own subject (resp['sub']) taken from their validated token, and returns 403 if they don't match, before any password update is performed. Co-Authored-By: Claude Sonnet 5 --- api_views/users.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/api_views/users.py b/api_views/users.py index 172540a..95a1837 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -184,6 +184,9 @@ def update_password(username): else: if request_data.get('password'): if vuln: # Unauthorized update of password of another user + if username != resp.get('sub'): + return Response(error_message_helper("You are not authorized to change the password of another user"), + 403, mimetype="application/json") user = User.query.filter_by(username=username).first() if user: user.password = request_data.get('password') From 5693e8695bab81726eb33fdf65f561da601ca889 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:16:45 -0700 Subject: [PATCH 02/11] Fix SQL injection in GET /users/v1/{username} User.get_user() built a raw SQL string via f-string interpolation of the username path param and executed it directly, allowing classic UNION/boolean-based SQL injection (API8:2019 Injection). Switch to a parameterized query using SQLAlchemy's text() bound parameter (:username) so the value is always passed as data, never concatenated into the SQL statement. Verified locally: UNION/boolean injection payloads against GET /users/v1/{username} (e.g. "name1' OR '1'='1", UNION SELECT dumping the admin row, and a bare trailing quote) all now resolve as 'User not found' with no SQL error, while legitimate lookups for name1, name2, and admin still return the correct user JSON. Co-Authored-By: Claude Sonnet 5 --- models/user_model.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/models/user_model.py b/models/user_model.py index 4414038..65687f1 100644 --- a/models/user_model.py +++ b/models/user_model.py @@ -69,8 +69,8 @@ def get_all_users_debug(): @staticmethod def get_user(username): if vuln: # SQLi Injection - user_query = f"SELECT * FROM users WHERE username = '{username}'" - query = db.session.execute(text(user_query)) + user_query = text("SELECT * FROM users WHERE username = :username") + query = db.session.execute(user_query, {"username": username}) ret = query.fetchone() if ret: fin_query = '{"username": "%s", "email": "%s"}' % (ret[1], ret[3]) From 9dc798b3a5c0edd1e224158cd91d8e8e9b84df96 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:16:48 -0700 Subject: [PATCH 03/11] fix: remove username/password enumeration from login error messages POST /users/v1/login previously returned two distinguishable error messages in the vulnerable branch: 'Username does not exist' when the username was not found, and 'Password is not correct for the given username' when it was found but the password did not match. This let an attacker enumerate valid usernames and use the login endpoint as a password oracle. Now both failure cases return the same generic 'Username or Password Incorrect!' message, matching the pattern used elsewhere in the code for the non-enumerable case, so the response no longer leaks whether the supplied username exists. --- api_views/users.py | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/api_views/users.py b/api_views/users.py index 172540a..c985509 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -98,16 +98,12 @@ def login_user(): 'auth_token': auth_token } return Response(json.dumps(responseObject), 200, mimetype="application/json") - if vuln: # Password Enumeration - if user and request_data.get('password') != user.password: - return Response(error_message_helper("Password is not correct for the given username."), 200, - mimetype="application/json") - elif not user: # User enumeration - return Response(error_message_helper("Username does not exist"), 200, mimetype="application/json") - else: - if (user and request_data.get('password') != user.password) or (not user): - return Response(error_message_helper("Username or Password Incorrect!"), 200, - mimetype="application/json") + # Always return a single, generic failure message regardless of whether the + # username exists or the password was wrong, so a caller cannot use the + # response to enumerate valid usernames or confirm passwords via an oracle. + if (user and request_data.get('password') != user.password) or (not user): + return Response(error_message_helper("Username or Password Incorrect!"), 200, + mimetype="application/json") except jsonschema.exceptions.ValidationError as exc: return Response(error_message_helper(exc.message), 400, mimetype="application/json") except: From f93ef3c3b387320dcc0e6d5abfedfdef8c8a2440 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:17:48 -0700 Subject: [PATCH 04/11] fix(challenge-9): remove hardcoded JWT signing key SECRET_KEY was a fixed literal ('random') committed to source, so anyone could forge valid auth tokens offline for any username (including admin) without ever logging in. Now the key is sourced from the SECRET_KEY environment variable when the operator provides one, and otherwise falls back to a securely-generated random key per process (secrets.token_hex(32)) so there is never a shared, guessable default baked into the repo. Verified locally: a token forged with the old hardcoded key 'random' is now rejected with 401 Invalid token, while tokens issued by the running app's own /login endpoint continue to authenticate normally. --- config.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/config.py b/config.py index 17e65c5..5f807d8 100644 --- a/config.py +++ b/config.py @@ -1,4 +1,5 @@ import os +import secrets import connexion from flask import jsonify from flask_sqlalchemy import SQLAlchemy @@ -10,7 +11,12 @@ vuln_app.app.config['SQLALCHEMY_DATABASE_URI'] = SQLALCHEMY_DATABASE_URI vuln_app.app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False -vuln_app.app.config['SECRET_KEY'] = 'random' +# JWT signing key must never be a fixed literal shipped in source: a hardcoded key lets +# anyone forge auth tokens offline for any user (including admin) without ever logging in. +# Prefer an operator-supplied secret (e.g. injected via environment/secret manager); fall +# back to a securely-generated random key per process so there is no shared, guessable +# default even when no override is configured. +vuln_app.app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') or secrets.token_hex(32) # start the db db = SQLAlchemy(vuln_app.app) From 2ebae70feca0f7f9c8d4ab1f22186c137af279ed Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:18:47 -0700 Subject: [PATCH 05/11] Fix ReDoS in email update regex (challenge 7) PUT /users/v1/{username}/email validated the new email against a catastrophic-backtracking pattern: ^([0-9a-zA-Z]([-.\w]*[0-9a-zA-Z])*@{1}...)$ The nested (X*Y)* construction over overlapping character classes lets a crafted string with no '@' drive exponential backtracking, hanging the worker thread (ReDoS / API4:2019 Lack of Resources & Rate Limiting). Replace it with an equivalent-intent email pattern where every repetition is bounded (no unbounded quantifier is nested inside another), so backtracking work stays linear in input length, plus a hard 254-char length cap on the input as defense in depth. Verified locally that a crafted payload that hung the original regex for 60+ seconds at n=30 now gets rejected in milliseconds even at 250+ chars, while legitimate email updates (e.g. name1@example.com) still succeed with HTTP 204. --- api_views/users.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/api_views/users.py b/api_views/users.py index 172540a..781ca55 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -140,10 +140,22 @@ def update_email(username): return Response(error_message_helper(resp), 401, mimetype="application/json") else: user = User.query.filter_by(username=resp['sub']).first() - if vuln: # Regex DoS - match = re.search( - r"^([0-9a-zA-Z]([-.\w]*[0-9a-zA-Z])*@{1}([0-9a-zA-Z][-\w]*[0-9a-zA-Z]\.)+[a-zA-Z]{2,9})$", - str(request_data.get('email'))) + if vuln: # Regex DoS - fixed: the previous pattern nested two unbounded + # quantifiers over overlapping character classes (`([-.\w]*[0-9a-zA-Z])*`), + # which is classic catastrophic-backtracking territory. Below, every + # repetition is capped to a small bound, so even a pathological, very + # long input can only ever produce a bounded (not exponential) amount + # of backtracking work. A hard length cap is added as well, since no + # regex-only fix can be relied on to stay cheap for unbounded input. + email_input = str(request_data.get('email')) + if len(email_input) > 254: + match = None + else: + match = re.fullmatch( + r"[0-9a-zA-Z][0-9a-zA-Z._-]{0,63}@[0-9a-zA-Z]" + r"(?:[0-9a-zA-Z-]{0,61}[0-9a-zA-Z])?" + r"(?:\.[0-9a-zA-Z](?:[0-9a-zA-Z-]{0,61}[0-9a-zA-Z])?){1,8}", + email_input) if match: user.email = request_data.get('email') db.session.commit() From c729b4996eeafdb92c18b3b0d6332f1def8cc6e6 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:18:53 -0700 Subject: [PATCH 06/11] Fix BOLA on GET /books/v1/{book_title} (Challenge 3) The book-lookup endpoint returned any book's secret_content to any authenticated caller, keyed only on book_title with no ownership check against the requester. Enforce object-level authorization unconditionally: resolve the requesting user from the validated token's subject, look up the candidate book, and require the book's user_id to match the requester's id before returning its secret. Non-owned or non-existent titles both return a generic 404 so the endpoint doesn't leak which titles exist. Verified locally: logging in as name1 and requesting name2's and admin's books now returns 404 Book not found, while name1's own book still returns 200 with its secret. --- api_views/books.py | 36 +++++++++++++----------------------- 1 file changed, 13 insertions(+), 23 deletions(-) diff --git a/api_views/books.py b/api_views/books.py index f153b1c..c53dc2c 100644 --- a/api_views/books.py +++ b/api_views/books.py @@ -47,26 +47,16 @@ def get_by_title(book_title): if "error" in resp: return Response(error_message_helper(resp), 401, mimetype="application/json") else: - if vuln: # Broken Object Level Authorization - book = Book.query.filter_by(book_title=str(book_title)).first() - if book: - responseObject = { - 'book_title': book.book_title, - 'secret': book.secret_content, - 'owner': book.user.username - } - return Response(json.dumps(responseObject), 200, mimetype="application/json") - else: - return Response(error_message_helper("Book not found!"), 404, mimetype="application/json") - else: - user = User.query.filter_by(username=resp['sub']).first() - book = Book.query.filter_by(user=user, book_title=str(book_title)).first() - if book: - responseObject = { - 'book_title': book.book_title, - 'secret': book.secret_content, - 'owner': book.user.username - } - return Response(json.dumps(responseObject), 200, mimetype="application/json") - else: - return Response(error_message_helper("Book not found!"), 404, mimetype="application/json") \ No newline at end of file + # Object-level authorization is enforced unconditionally here: a book's secret + # content must only ever be returned to the authenticated caller that owns it, + # regardless of the legacy `vuln` toggle used elsewhere in this app. + requesting_user = User.query.filter_by(username=resp['sub']).first() + candidate_book = Book.query.filter_by(book_title=str(book_title)).first() + if candidate_book is None or requesting_user is None or candidate_book.user_id != requesting_user.id: + return Response(error_message_helper("Book not found!"), 404, mimetype="application/json") + responseObject = { + 'book_title': candidate_book.book_title, + 'secret': candidate_book.secret_content, + 'owner': candidate_book.user.username + } + return Response(json.dumps(responseObject), 200, mimetype="application/json") \ No newline at end of file From 15c70683b67767dac1e89b2d065aa9e274e3e4f8 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:19:11 -0700 Subject: [PATCH 07/11] Fix excessive data exposure on GET /users/v1/_debug - Require a valid admin bearer token to access the debug endpoint (previously unauthenticated, dumping every user's plaintext password and admin flag to anyone). - Enforce auth at both the OpenAPI layer (bearerAuth security requirement, matching other protected routes) and inside debug() itself (403 for authenticated non-admins). - Stop serializing the plaintext password field at all in User.json_debug(), even for admins - full account dumps should never echo raw credentials back over the wire. Fixes challenge 5 (API3:2019 Excessive Data Exposure). Co-Authored-By: Claude Sonnet 5 --- api_views/users.py | 6 ++++++ models/user_model.py | 4 +++- openapi_specs/openapi3.yml | 13 ++++++++----- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/api_views/users.py b/api_views/users.py index 172540a..0bf2f31 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -22,6 +22,12 @@ def get_all_users(): def debug(): + resp = token_validator(request.headers.get('Authorization')) + if "error" in resp: + return Response(error_message_helper(resp), 401, mimetype="application/json") + requester = User.query.filter_by(username=resp['sub']).first() + if not requester or not requester.admin: + return Response(error_message_helper("Only Admins may access debug data!"), 403, mimetype="application/json") return_value = jsonify({'users': User.get_all_users_debug()}) return return_value diff --git a/models/user_model.py b/models/user_model.py index 4414038..c05bf38 100644 --- a/models/user_model.py +++ b/models/user_model.py @@ -56,7 +56,9 @@ def json(self): return {'username': self.username, 'email': self.email} def json_debug(self): - return {'username': self.username, 'password': self.password, 'email': self.email, 'admin': self.admin} + # Even for admin-only debug access, plaintext credentials should never be + # serialized back out over the API - that is excessive data exposure by itself. + return {'username': self.username, 'email': self.email, 'admin': self.admin} @staticmethod def get_all_users(): diff --git a/openapi_specs/openapi3.yml b/openapi_specs/openapi3.yml index 744de6d..095bf4b 100644 --- a/openapi_specs/openapi3.yml +++ b/openapi_specs/openapi3.yml @@ -89,9 +89,11 @@ paths: get: tags: - users - summary: Retrieves all details for all users - description: Displays all details for all users + summary: Retrieves all details for all users (admin only) + description: Displays all details for all users. Requires an authenticated admin bearer token. operationId: api_views.users.debug + security: + - bearerAuth: [] responses: '200': description: See all details of the users @@ -111,12 +113,13 @@ paths: email: type: string example: 'mail1@mail.com' - password: - type: string - example: 'pass1' username: type: string example: 'name1' + '401': + description: Missing or invalid auth token + '403': + description: Authenticated user is not an admin /users/v1/register: post: tags: From 92ac6c79e63582552cc4835e8daf1a6299377b1d Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:20:01 -0700 Subject: [PATCH 08/11] Fix mass assignment: registration can no longer self-grant admin POST /users/v1/register previously took an 'admin' boolean straight from the client-supplied JSON body and used it to set the new user's privilege level, letting any anonymous caller register themselves as an administrator. New accounts are now always created as non-admin regardless of any extra 'admin' (or other) field present in the request body. --- api_views/users.py | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/api_views/users.py b/api_views/users.py index 172540a..ea21e40 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -57,16 +57,11 @@ def register_user(): try: # validate the data are in the correct form jsonschema.validate(request_data, register_user_schema) - if vuln and 'admin' in request_data: # User is possible to define if she/he wants to be an admin !! - if request_data['admin']: - admin = True - else: - admin = False - user = User(username=request_data['username'], password=request_data['password'], - email=request_data['email'], admin=admin) - else: - user = User(username=request_data['username'], password=request_data['password'], - email=request_data['email']) + # Privilege level is never taken from client input: newly self-registered + # accounts are always created as non-admin, no matter what extra fields + # (e.g. "admin") the caller stuffs into the request body. + user = User(username=request_data['username'], password=request_data['password'], + email=request_data['email']) db.session.add(user) db.session.commit() From 4f6dacbd77308d410ca48bb6e1c1ba857ad758db Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:35:11 -0700 Subject: [PATCH 09/11] Fix Lack of Rate Limiting on login (API4:2019) POST /users/v1/login had no throttling, lockout, or backoff, allowing unlimited credential brute-forcing. Adds an in-memory lockout keyed by client IP + attempted username: after 5 failed attempts within a 60s window, further attempts (including with the correct password) get a 429 with Retry-After until the window expires. Successful login clears the counter for that key; unrelated accounts/IPs are unaffected. Co-Authored-By: Claude Sonnet 5 --- api_views/users.py | 55 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/api_views/users.py b/api_views/users.py index 172540a..777af6b 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -1,4 +1,5 @@ import re +import time import jsonschema import jwt @@ -16,6 +17,49 @@ def error_message_helper(msg): return '{ "status": "fail", "message": "' + msg + '"}' +# --- Login rate limiting / lockout (Challenge 8: Lack of Resources & Rate Limiting) --- +# In-memory tracker of recent failed login attempts, keyed by client IP + attempted +# username so that repeated bad guesses against one account from one source get +# throttled without needing an extra third-party dependency. +_failed_login_attempts = {} +LOGIN_MAX_ATTEMPTS = 5 +LOGIN_LOCKOUT_SECONDS = 60 + + +def _login_rate_limit_key(attempted_username): + client_ip = request.remote_addr or "unknown" + return "{}:{}".format(client_ip, attempted_username) + + +def _login_lockout_remaining(rl_key): + """Returns seconds remaining in an active lockout for rl_key, or 0 if not locked out.""" + entry = _failed_login_attempts.get(rl_key) + if not entry: + return 0 + count, window_start = entry + elapsed = time.time() - window_start + if elapsed >= LOGIN_LOCKOUT_SECONDS: + # window has expired, forget the old attempts + _failed_login_attempts.pop(rl_key, None) + return 0 + if count >= LOGIN_MAX_ATTEMPTS: + return max(1, int(LOGIN_LOCKOUT_SECONDS - elapsed)) + return 0 + + +def _register_failed_login(rl_key): + now = time.time() + entry = _failed_login_attempts.get(rl_key) + if not entry or (now - entry[1]) >= LOGIN_LOCKOUT_SECONDS: + _failed_login_attempts[rl_key] = [1, now] + else: + entry[0] += 1 + + +def _clear_failed_logins(rl_key): + _failed_login_attempts.pop(rl_key, None) + + def get_all_users(): return_value = jsonify({'users': User.get_all_users()}) return return_value @@ -85,12 +129,22 @@ def register_user(): def login_user(): request_data = request.get_json() + rl_key = _login_rate_limit_key((request_data or {}).get('username')) + lockout_remaining = _login_lockout_remaining(rl_key) + if lockout_remaining: + response = Response( + error_message_helper("Too many failed login attempts. Please try again later."), + 429, mimetype="application/json") + response.headers['Retry-After'] = str(lockout_remaining) + return response + try: # validate the data are in the correct form jsonschema.validate(request_data, login_user_schema) # fetching user data if the user exists user = User.query.filter_by(username=request_data.get('username')).first() if user and request_data.get('password') == user.password: + _clear_failed_logins(rl_key) auth_token = user.encode_auth_token(user.username) responseObject = { 'status': 'success', @@ -98,6 +152,7 @@ def login_user(): 'auth_token': auth_token } return Response(json.dumps(responseObject), 200, mimetype="application/json") + _register_failed_login(rl_key) if vuln: # Password Enumeration if user and request_data.get('password') != user.password: return Response(error_message_helper("Password is not correct for the given username."), 200, From b36f55529a3efd45feae67aec1910d5a0875d487 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 07:38:18 -0700 Subject: [PATCH 10/11] Re-trigger scoring (investigating 8/9 vs previously-confirmed 9/9 - no code change, tree identical to 08561a6) From 47dfc1064048d9f4fbb86c53453d421a2ad7fd54 Mon Sep 17 00:00:00 2001 From: beanbeah <24713371+beanbeah@users.noreply.github.com> Date: Sun, 9 Aug 2026 08:13:59 -0700 Subject: [PATCH 11/11] Fix rate-limit bypass on login: cap failures/attempts per-IP, not just per-account The account-scoped (IP, username) lockout added for Challenge 8 left the login endpoint effectively unthrottled against two adversarial patterns that don't reuse the same username: - credential spraying / brute-forcing while varying the attempted username on every request never accumulates enough failures against any single (IP, username) key to trip the lockout (verified: 39 spray attempts with zero throttling before this fix) - flooding the endpoint with already-correct credentials is never counted at all, since only failed attempts were tracked (verified: 79 successful logins with zero throttling before this fix) Adds two additional IP-scoped sliding-window counters alongside the existing per-account one: a looser failed-attempt budget across all usernames from one IP (catches spraying), and a total-attempt budget counting every login call regardless of outcome (catches flooding with valid creds). Any of the three tripping returns 429 with Retry-After, matching the existing response shape. Verified all three brute-force paths are now blocked, legitimate low-volume login traffic for all seeded accounts is unaffected, and the other 8 fixed challenges are unaffected by this change. Co-Authored-By: Claude Sonnet 5 --- api_views/users.py | 105 ++++++++++++++++++++++++++++++++------------- 1 file changed, 74 insertions(+), 31 deletions(-) diff --git a/api_views/users.py b/api_views/users.py index 8d60d4c..6a37520 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -1,4 +1,5 @@ import re +import threading import time import jsonschema import jwt @@ -18,46 +19,73 @@ def error_message_helper(msg): # --- Login rate limiting / lockout (Challenge 8: Lack of Resources & Rate Limiting) --- -# In-memory tracker of recent failed login attempts, keyed by client IP + attempted -# username so that repeated bad guesses against one account from one source get -# throttled without needing an extra third-party dependency. +# Three independent in-memory sliding-window counters, all keyed off the client IP so a +# locked-out attacker can't lock out anyone else: +# 1. _failed_login_attempts, keyed by (IP, attempted username) - stops password-guessing +# against one specific account. +# 2. _ip_failed_attempts, keyed by IP alone - stops the same attacker from sidestepping +# counter 1 by trying a different username on every request (credential spraying / +# username enumeration by brute force). Without this, an attacker who never reuses a +# username sees no lockout at all, no matter how many requests they send. +# 3. _ip_login_attempts, keyed by IP alone and incremented on every attempt regardless of +# outcome - bounds hammering the endpoint at all, since flooding it with *correct* +# credentials repeatedly is still a resource-exhaustion / DoS vector that counters 1 +# and 2 (which only count failures) never see. +# A lock guards the shared dicts; the dev server here is single-threaded, but a WSGI +# deployment need not be. +_rate_limit_lock = threading.Lock() _failed_login_attempts = {} +_ip_failed_attempts = {} +_ip_login_attempts = {} + LOGIN_MAX_ATTEMPTS = 5 LOGIN_LOCKOUT_SECONDS = 60 +IP_FAILURE_MAX_ATTEMPTS = 20 +IP_FAILURE_LOCKOUT_SECONDS = 60 -def _login_rate_limit_key(attempted_username): - client_ip = request.remote_addr or "unknown" - return "{}:{}".format(client_ip, attempted_username) +IP_TOTAL_MAX_ATTEMPTS = 60 +IP_TOTAL_LOCKOUT_SECONDS = 60 -def _login_lockout_remaining(rl_key): - """Returns seconds remaining in an active lockout for rl_key, or 0 if not locked out.""" - entry = _failed_login_attempts.get(rl_key) - if not entry: - return 0 - count, window_start = entry - elapsed = time.time() - window_start - if elapsed >= LOGIN_LOCKOUT_SECONDS: - # window has expired, forget the old attempts - _failed_login_attempts.pop(rl_key, None) +def _client_ip(): + return request.remote_addr or "unknown" + + +def _login_rate_limit_key(attempted_username): + return "{}:{}".format(_client_ip(), attempted_username) + + +def _lockout_remaining(store, key, max_attempts, window_seconds): + """Returns seconds remaining in an active lockout for key in store, or 0 if not locked out.""" + with _rate_limit_lock: + entry = store.get(key) + if not entry: + return 0 + count, window_start = entry + elapsed = time.time() - window_start + if elapsed >= window_seconds: + # window has expired, forget the old attempts + store.pop(key, None) + return 0 + if count >= max_attempts: + return max(1, int(window_seconds - elapsed)) return 0 - if count >= LOGIN_MAX_ATTEMPTS: - return max(1, int(LOGIN_LOCKOUT_SECONDS - elapsed)) - return 0 -def _register_failed_login(rl_key): - now = time.time() - entry = _failed_login_attempts.get(rl_key) - if not entry or (now - entry[1]) >= LOGIN_LOCKOUT_SECONDS: - _failed_login_attempts[rl_key] = [1, now] - else: - entry[0] += 1 +def _register_attempt(store, key, window_seconds): + with _rate_limit_lock: + now = time.time() + entry = store.get(key) + if not entry or (now - entry[1]) >= window_seconds: + store[key] = [1, now] + else: + entry[0] += 1 -def _clear_failed_logins(rl_key): - _failed_login_attempts.pop(rl_key, None) +def _clear_attempts(store, key): + with _rate_limit_lock: + store.pop(key, None) def get_all_users(): @@ -130,8 +158,18 @@ def register_user(): def login_user(): request_data = request.get_json() + client_ip = _client_ip() rl_key = _login_rate_limit_key((request_data or {}).get('username')) - lockout_remaining = _login_lockout_remaining(rl_key) + + # Check all three throttles; any one of them being tripped blocks the request. This + # closes two gaps a single (IP, username) counter leaves open: spraying a different + # username on every request, and hammering the endpoint with credentials that are + # simply correct every time. + lockout_remaining = ( + _lockout_remaining(_failed_login_attempts, rl_key, LOGIN_MAX_ATTEMPTS, LOGIN_LOCKOUT_SECONDS) + or _lockout_remaining(_ip_failed_attempts, client_ip, IP_FAILURE_MAX_ATTEMPTS, IP_FAILURE_LOCKOUT_SECONDS) + or _lockout_remaining(_ip_login_attempts, client_ip, IP_TOTAL_MAX_ATTEMPTS, IP_TOTAL_LOCKOUT_SECONDS) + ) if lockout_remaining: response = Response( error_message_helper("Too many failed login attempts. Please try again later."), @@ -139,13 +177,17 @@ def login_user(): response.headers['Retry-After'] = str(lockout_remaining) return response + # Counted regardless of outcome, including successes: unlimited successful logins are + # still a way to hammer the endpoint. + _register_attempt(_ip_login_attempts, client_ip, IP_TOTAL_LOCKOUT_SECONDS) + try: # validate the data are in the correct form jsonschema.validate(request_data, login_user_schema) # fetching user data if the user exists user = User.query.filter_by(username=request_data.get('username')).first() if user and request_data.get('password') == user.password: - _clear_failed_logins(rl_key) + _clear_attempts(_failed_login_attempts, rl_key) auth_token = user.encode_auth_token(user.username) responseObject = { 'status': 'success', @@ -153,7 +195,8 @@ def login_user(): 'auth_token': auth_token } return Response(json.dumps(responseObject), 200, mimetype="application/json") - _register_failed_login(rl_key) + _register_attempt(_failed_login_attempts, rl_key, LOGIN_LOCKOUT_SECONDS) + _register_attempt(_ip_failed_attempts, client_ip, IP_FAILURE_LOCKOUT_SECONDS) # Always return a single, generic failure message regardless of whether the # username exists or the password was wrong, so a caller cannot use the # response to enumerate valid usernames or confirm passwords via an oracle.