Revert "Change JWT encryption from HMAC to RSA with rotating keys (#37)"

This reverts commit 5079f28.

Author: Vianpyro

.devcontainer/Dockerfile

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

.github/workflows/pytest.yml

@@ -33,16 +33,6 @@ 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,8 +2,7 @@
 logs/
 *.log
 
-# Ignore JWT keys
-keys/
+# Ignore all pem files
 *.pem
 
 # Upload folder

Dockerfile

@@ -4,26 +4,23 @@ FROM python:3.13-slim
 # Set a specific working directory in the container
 WORKDIR /app
 
-# 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
+# Install dependencies separately for better caching
 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
 
-ENTRYPOINT ["python3", "app.py"]
+# Command to run the app
+CMD ["python", "app.py"]

app.py

@@ -1,18 +1,15 @@
 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)
@@ -29,10 +26,6 @@
 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

@@ -12,12 +12,12 @@ def setup_logging():
     logger.setLevel(logging.DEBUG)
 
     # Create a file handler that rotates logs daily
-    current_time = datetime.datetime.now().strftime("%Y-%m-%d")
+    timestamp = 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}/{current_time}.log"
+    log_filename = f"{LOGS_DIRECTORY}/{timestamp}.log"
 
     # Set up timed rotating file handler
     file_handler = TimedRotatingFileHandler(

config/logging.py

@@ -20,6 +20,7 @@ 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)
 

config/settings.py

@@ -0,0 +1,78 @@
+import os
+from datetime import datetime, timedelta, timezone
+from functools import wraps
+
+import jwt
+from flask import jsonify, request
+
+JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "SuperSecretKey")
+JWT_ACCESS_TOKEN_EXPIRY = timedelta(hours=1)
+JWT_REFRESH_TOKEN_EXPIRY = timedelta(days=30)
+
+
+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
+
+
+def generate_access_token(person_id: int) -> str:
+    """Generate a short-lived JWT access token for a user."""
+    payload = {
+        "person_id": person_id,
+        "exp": datetime.now(timezone.utc) + JWT_ACCESS_TOKEN_EXPIRY,  # Expiration
+        "iat": datetime.now(timezone.utc),  # Issued at
+        "token_type": "access",
+    }
+    return jwt.encode(payload, JWT_SECRET_KEY, algorithm="HS256")
+
+
+def generate_refresh_token(person_id: int) -> str:
+    """Generate a long-lived refresh token for a user."""
+    payload = {
+        "person_id": person_id,
+        "exp": datetime.now(timezone.utc) + JWT_REFRESH_TOKEN_EXPIRY,
+        "iat": datetime.now(timezone.utc),
+        "token_type": "refresh",
+    }
+    return jwt.encode(payload, JWT_SECRET_KEY, algorithm="HS256")
+
+
+def extract_token_from_header() -> str:
+    """Extract the Bearer token from the Authorization header."""
+    auth_header = request.headers.get("Authorization")
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise TokenError("Token is missing or improperly formatted", 401)
+    return auth_header.split("Bearer ")[1]
+
+
+def verify_token(token: str, required_type: str) -> dict:
+    """Verify and decode a JWT token."""
+    try:
+        decoded = jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS256"])
+        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)
+
+
+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