Change JWT encryption from HMAC to RSA with rotating keys

.devcontainer/Dockerfile

@@ -2,7 +2,7 @@ FROM alpine:latest
 
 # Install common tools
 RUN apk add --no-cache bash git \
-    python3 py3-pip
+    python3 py3-pip openssl
 
 # Setup default user
 ARG USERNAME=vscode

.github/workflows/pytest.yml

@@ -33,6 +33,16 @@ jobs:
           python-version: ${{ matrix.python-version }}
           cache: "pip"
 
+      - name: Install OpenSSL (Linux)
+        if: runner.os == 'Linux'
+        run: sudo apt-get update && sudo apt-get install -y libssl-dev
+
+      # MacOS does not require OpenSSL installation as it is pre-installed
+
+      - name: Install OpenSSL (Windows)
+        if: runner.os == 'Windows'
+        run: choco install openssl.light --no-progress
+
       - name: Install dependencies
         run: |
           pip install -r requirements.txt

.gitignore

@@ -2,7 +2,8 @@
 logs/
 *.log
 
-# Ignore all pem files
+# Ignore JWT keys
+keys/
 *.pem
 
 # Upload folder

Dockerfile

@@ -4,23 +4,26 @@ FROM python:3.13-slim
 # Set a specific working directory in the container
 WORKDIR /app
 
-# Install dependencies separately for better caching
+# Install dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends curl openssl \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+# Install required Python packages
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
 
 # Copy the rest of the application files
 COPY . .
 
-# Expose the Flask port
-EXPOSE 5000
-
 # Create a non-root user and set permissions for the /app directory
 RUN adduser --disabled-password --gecos '' apiuser && chown -R apiuser /app
 USER apiuser
 
+# Expose the Flask port
+EXPOSE 5000
+
 # Add health check for the container
 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
     CMD curl --fail http://localhost:5000/ || exit 1
 
-# Command to run the app
-CMD ["python", "app.py"]
+ENTRYPOINT ["python3", "app.py"]

app.py

@@ -1,15 +1,18 @@
 import argparse
+import os
 import traceback
 
 from flask import Flask, jsonify, request
 from flask_cors import CORS
 from werkzeug.exceptions import HTTPException
 
+from config.jwtoken import ACTIVE_KID_FILE
 from config.logging import setup_logging
 from config.ratelimit import limiter
 from config.settings import Config
 from routes import register_routes
 from utility.database import extract_error_message
+from utility.jwtoken.keys_rotation import rotate_keys
 
 app = Flask(__name__)
 app.config.from_object(Config)
@@ -26,6 +29,10 @@
 if app.config["TESTING"]:
     limiter.enabled = False
 
+# Ensure the keys directory and active_kid.txt file exist
+if not os.path.exists(ACTIVE_KID_FILE):
+    rotate_keys()
+
 
 @app.route("/")
 def home():

config/jwtoken.py

@@ -0,0 +1,8 @@
+from pathlib import Path
+
+KEYS_DIR = Path("keys")
+
+ACTIVE_KID_FILE = KEYS_DIR / "active_kid.txt"
+CREATED_AT_FILE = "created_at.txt"
+PRIVATE_KEY_FILE = "private.pem"
+PUBLIC_KEY_FILE = "public.pem"

config/logging.py

@@ -12,12 +12,12 @@ def setup_logging():
     logger.setLevel(logging.DEBUG)
 
     # Create a file handler that rotates logs daily
-    timestamp = datetime.datetime.now().strftime("%Y-%m-%d")
+    current_time = datetime.datetime.now().strftime("%Y-%m-%d")
 
     # Create a directory for logs if it doesn't exist
     if not os.path.exists(LOGS_DIRECTORY):
         os.makedirs(LOGS_DIRECTORY)
-    log_filename = f"{LOGS_DIRECTORY}/{timestamp}.log"
+    log_filename = f"{LOGS_DIRECTORY}/{current_time}.log"
 
     # Set up timed rotating file handler
     file_handler = TimedRotatingFileHandler(

config/settings.py

@@ -20,7 +20,6 @@ class Config:
     MYSQL_CURSORCLASS = "DictCursor"
 
     # JWT configuration
-    JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY")
     JWT_ACCESS_TOKEN_EXPIRY = timedelta(hours=1)
     JWT_REFRESH_TOKEN_EXPIRY = timedelta(days=30)
 

jwtoken/decorators.py

@@ -0,0 +1,24 @@
+from functools import wraps
+
+from flask import jsonify, request
+
+from utility.jwtoken.common import extract_token_from_header
+
+from .exceptions import TokenError
+from .tokens import verify_token
+
+
+def token_required(f):
+    """Decorator to protect routes by requiring a valid token."""
+
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        try:
+            token = extract_token_from_header()
+            decoded = verify_token(token, required_type="access")
+            request.person_id = decoded["person_id"]
+            return f(*args, **kwargs)
+        except TokenError as e:
+            return jsonify(message=e.message), e.status_code
+
+    return decorated

jwtoken/exceptions.py

@@ -0,0 +1,7 @@
+class TokenError(Exception):
+    """Custom exception for token-related errors."""
+
+    def __init__(self, message, status_code):
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message

jwtoken/tokens.py

@@ -0,0 +1,63 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+
+from utility.jwtoken.common import get_active_kid, load_private_key, load_public_key
+
+from .exceptions import TokenError
+
+JWT_ACCESS_TOKEN_EXPIRY = timedelta(hours=1)
+JWT_REFRESH_TOKEN_EXPIRY = timedelta(days=30)
+
+
+def generate_access_token(person_id: int) -> str:
+    kid = get_active_kid()
+    private_key = load_private_key(kid)
+
+    current_time = datetime.now(timezone.utc)
+    payload = {
+        "person_id": person_id,
+        "exp": current_time + JWT_ACCESS_TOKEN_EXPIRY,  # Expiration
+        "iat": current_time,  # Issued at
+        "token_type": "access",
+    }
+    headers = {"kid": kid}
+    return jwt.encode(payload, private_key, algorithm="RS256", headers=headers)
+
+
+def generate_refresh_token(person_id: int) -> str:
+    """Generate a long-lived refresh token for a user."""
+    kid = get_active_kid()
+    private_key = load_private_key(kid)
+
+    current_time = datetime.now(timezone.utc)
+    payload = {
+        "person_id": person_id,
+        "exp": current_time + JWT_REFRESH_TOKEN_EXPIRY,  # Expiration
+        "iat": current_time,  # Issued at
+        "token_type": "refresh",
+    }
+    headers = {"kid": kid}
+
+    return jwt.encode(payload, private_key, algorithm="RS256", headers=headers)
+
+
+def verify_token(token: str, required_type: str) -> dict:
+    """Verify and decode a JWT token."""
+    try:
+        unverified_header = jwt.get_unverified_header(token)
+        kid = unverified_header.get("kid")
+        if not kid:
+            raise TokenError("KID missing in token header", 401)
+
+        public_key = load_public_key(kid)
+
+        decoded = jwt.decode(token, public_key, algorithms=["RS256"])
+        if decoded.get("token_type") != required_type:
+            raise jwt.InvalidTokenError("Invalid token type")
+        return decoded
+
+    except jwt.ExpiredSignatureError:
+        raise TokenError("Token has expired", 401)
+    except jwt.InvalidTokenError:
+        raise TokenError("Invalid token", 401)

routes/authentication.py

@@ -3,15 +3,11 @@
 from pymysql import MySQLError
 
 from config.ratelimit import limiter
-from jwt_helper import (
-    TokenError,
-    extract_token_from_header,
-    generate_access_token,
-    generate_refresh_token,
-    verify_token,
-)
+from jwtoken.exceptions import TokenError
+from jwtoken.tokens import generate_access_token, generate_refresh_token, verify_token
 from utility.database import database_cursor
 from utility.encryption import encrypt_email, hash_email, hash_password, verify_password
+from utility.jwtoken.common import extract_token_from_header
 from utility.validation import validate_email, validate_password
 
 authentication_blueprint = Blueprint("authentication", __name__)

routes/picture.py

@@ -3,7 +3,7 @@
 from flask import Blueprint, jsonify, request, send_from_directory
 
 from config.settings import Config
-from jwt_helper import token_required
+from jwtoken.decorators import token_required
 from utility.database import database_cursor
 
 picture_blueprint = Blueprint("picture", __name__)