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 diff --git a/api_views/users.py b/api_views/users.py index 172540a..6a37520 100644 --- a/api_views/users.py +++ b/api_views/users.py @@ -1,4 +1,6 @@ import re +import threading +import time import jsonschema import jwt @@ -16,12 +18,88 @@ def error_message_helper(msg): return '{ "status": "fail", "message": "' + msg + '"}' +# --- Login rate limiting / lockout (Challenge 8: Lack of Resources & Rate Limiting) --- +# 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 + +IP_TOTAL_MAX_ATTEMPTS = 60 +IP_TOTAL_LOCKOUT_SECONDS = 60 + + +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 + + +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_attempts(store, key): + with _rate_limit_lock: + store.pop(key, None) + + def get_all_users(): return_value = jsonify({'users': User.get_all_users()}) return return_value 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 @@ -57,16 +135,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() @@ -85,12 +158,36 @@ 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')) + + # 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."), + 429, mimetype="application/json") + 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_attempts(_failed_login_attempts, rl_key) auth_token = user.encode_auth_token(user.username) responseObject = { 'status': 'success', @@ -98,16 +195,14 @@ 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") + _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. + 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: @@ -140,10 +235,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() @@ -184,6 +291,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') 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) diff --git a/models/user_model.py b/models/user_model.py index 4414038..7ff84e6 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(): @@ -69,8 +71,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]) 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: