Fix all validated VAmPI CTF challenges

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")
+        # 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")

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,29 +158,51 @@ 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',
                 'message': 'Successfully logged in.',
                 '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')

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)
 

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])

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: