Secure routes with write access to database

app/__init__.py

@@ -2,22 +2,23 @@
 
 import os
 
-from dotenv import load_dotenv
 from flask import Flask
 
+from app.config import Config
 
-def create_app():
+
+def create_app(test_config=None):
     """Application factory pattern."""
-    # Load the env variables to use here
-    load_dotenv()
 
     app = Flask(__name__)
-    # Use app.config to set config connection details
-    app.config["MONGO_URI"] = os.getenv("MONGO_CONNECTION")
-    app.config["DB_NAME"] = os.getenv("PROJECT_DATABASE")
-    app.config["COLLECTION_NAME"] = os.getenv("PROJECT_COLLECTION")
 
-    # Import routes — routes can import app safely because it exists
+    # 1. Load the default configuration from the Config object.
+    app.config.from_object(Config)
+
+    if test_config:  # Override with test specifics
+        app.config.from_mapping(test_config)
+
+    # Import routes
     from app.routes import \
         register_routes  # pylint: disable=import-outside-toplevel
 

app/config.py

@@ -0,0 +1,34 @@
+# pylint: disable=too-few-public-methods
+
+"""
+Application configuration module for Flask.
+
+Loads environment variables from a .env file and defines the Config class
+used to configure Flask and database connection settings.
+"""
+
+import os
+
+from dotenv import load_dotenv
+
+# Find the absolute path of the root directory of the project
+basedir = os.path.abspath(os.path.dirname(__file__))
+# Build the path to the .env file located in the parent directory of the project root
+dotenv_path = os.path.join(os.path.dirname(basedir), ".env")
+# Load environment variables to be used
+load_dotenv(dotenv_path=dotenv_path)
+
+
+class Config:
+    """Set Flask configuration variables from environment variables"""
+
+    # General config
+    SECRET_KEY = os.environ.get("SECRET_KEY")
+    FLASK_APP = os.environ.get("FLASK_APP")
+    FLASK_ENV = os.environ.get("FLASK")  # values will be 'development' or 'production'
+    API_KEY = os.environ.get("API_KEY")
+
+    # Database config
+    MONGO_URI = os.environ.get("MONGO_CONNECTION")
+    DB_NAME = os.environ.get("PROJECT_DATABASE")
+    COLLECTION_NAME = os.environ.get("PROJECT_COLLECTION")

app/routes.py

@@ -4,15 +4,15 @@
 import uuid
 
 from flask import jsonify, request
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import HTTPException, NotFound
 
 from app.datastore.mongo_db import get_book_collection
 from app.datastore.mongo_helper import insert_book_to_mongo
+from app.utils.api_security import require_api_key
 from app.utils.helper import append_hostname
 from data import books
 
 
-# ----------- POST section ------------------
 def register_routes(app):  # pylint: disable=too-many-statements
     """
     Register all Flask routes with the given app instance.
@@ -21,7 +21,9 @@ def register_routes(app):  # pylint: disable=too-many-statements
         app (Flask): The Flask application instance to register routes on.
     """
 
+    # ----------- POST section ------------------
     @app.route("/books", methods=["POST"])
+    @require_api_key
     def add_book():
         """Function to add a new book to the collection."""
         # check if request is json
@@ -149,6 +151,7 @@ def get_book(book_id):
 
     # ----------- DELETE section ------------------
     @app.route("/books/<string:book_id>", methods=["DELETE"])
+    @require_api_key
     def delete_book(book_id):
         """
         Soft delete a book by setting its state to 'deleted' or return error if not found.
@@ -165,6 +168,7 @@ def delete_book(book_id):
     # ----------- PUT section ------------------
 
     @app.route("/books/<string:book_id>", methods=["PUT"])
+    @require_api_key
     def update_book(book_id):
         """
         Update a book by its unique ID using JSON from the request body.
@@ -215,11 +219,30 @@ def update_book(book_id):
 
         return jsonify({"error": "Book not found"}), 404
 
+
+    # ----------- CUSTOM ERROR HANDLERS ------------------
+
     @app.errorhandler(NotFound)
     def handle_not_found(e):
         """Return a custom JSON response for 404 Not Found errors."""
         return jsonify({"error": str(e)}), 404
 
+    @app.errorhandler(HTTPException)
+    def handle_http_exception(e):
+        """Return JSON for any HTTPException (401, 404, 403, etc.)
+        preserving its original status code & description.
+        """
+        # e.code is the HTTP status code (e.g. 401)
+        # e.description is the text you passed to abort()
+        response = {
+            "error": {
+                "code": e.code,
+                "name": e.name,
+                "message": e.description
+            }
+        }
+        return jsonify(response), e.code
+
     @app.errorhandler(Exception)
     def handle_exception(e):
         """Return a custom JSON response for any exception."""

app/utils/api_security.py

@@ -0,0 +1,47 @@
+"""API security decorators."""
+
+import hmac
+from functools import wraps
+
+from flask import abort, current_app, request
+
+
+def log_unauthorized_access():
+    """
+    Helper function to log a warning for unauthorized access attempts with IP and path info.
+    """
+    current_app.logger.warning(
+        f"Unauthorized access attempt: IP={request.remote_addr}, path={request.path}"
+    )
+
+
+def require_api_key(f):
+    """A decorator to protect routes with a required API key"""
+
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+
+        expected_key = current_app.config.get("API_KEY")
+
+        if not expected_key:
+            # Secure logging — don't leak keys
+            log_unauthorized_access()
+
+            abort(500, description="API key not configured on the server.")
+
+        provided_key = request.headers.get("X-API-KEY")
+        if not provided_key:
+            # Secure logging — don't leak keys
+            log_unauthorized_access()
+            abort(401, description="API key is missing.")
+
+        # Securely compare the provided key with the expected key
+        if not hmac.compare_digest(provided_key, expected_key):
+            # Secure logging — don't leak keys
+            log_unauthorized_access()
+            abort(401, description="Invalid API key.")
+
+        # If key is valid, proceed with the original function
+        return f(*args, **kwargs)
+
+    return decorated_function