feat: basic api-key based authentication

src/config.ini

@@ -9,17 +9,17 @@ peer_keep_alive = 21
 [Server]
 hostname = 0.0.0.0
 port = 10086
-debug_enabled = true
+debug_enabled = True
 wg_conf_path = /etc/wireguard
 awg_conf_path = /etc/amnezia/amneziawg
 app_prefix =
-auth_req = true
+auth_req = True
 version = v5.0.0
 dashboard_refresh_interval = 60000
 dashboard_peer_list_display = grid
 dashboard_sort = status
 dashboard_theme = dark
-wgdashboard_apikey = false
+wgdashboard_apikey = true
 dashboard_language = en-US
 log_level = DEBUG
 
@@ -31,6 +31,7 @@ totp_verified = false
 totp_key = ZSI2Z4QHSGVAK6TFVMZXORFDWKHDN4TQ
 
 [Other]
+welcome_session = false
 
 [Database]
 type = sqlite

src/main.py

@@ -6,6 +6,7 @@ from logging.config import dictConfig
 import flask
 import json
 import os
+import secrets
 
 from modules.config.reader import reader
 from modules.database.database import database
@@ -35,11 +36,17 @@ if __name__ == '__main__':
     ok, engine, session = database.create_session(config_database)
     ok = database.ensure_contents(engine)
 
+    prefix = config_server.get('app_prefix', '')
+
     # Configure the Flask app
     app = flask.Flask("WGDashboard", template_folder=os.path.abspath("./static/dist/WGDashboardAdmin"))
-    app.register_blueprint(routes)
+    app.register_blueprint(routes, url_prefix=prefix)
 
     app.wgd_config = config_contents
+
+    app.secret_key = secrets.token_urlsafe(64)
+    app.config['SESSION_TYPE'] = 'filesystem'
+
     app.engine = engine
     app.db_session = session
 
@@ -50,4 +57,5 @@ if __name__ == '__main__':
         debug=debug_enabled,
         host=hostname,
         port=port,
-        use_reloader=False)
+        use_reloader=False
+    )

src/modules/config/utilities.py

@@ -49,6 +49,15 @@ class checks():
             config_dict = {}
             for section in config.sections():
                 items = dict(config.items(section))
+
+                for key, value in items.items():
+                    if value.strip().lower() == 'true':
+                        items[key] = True
+                    elif value.strip().lower() == 'false':
+                        items[key] = False
+                    else:
+                        items[key] = value
+
                 config_dict[section] = items
 
             return True, config_dict

src/modules/database/functions.py

@@ -3,14 +3,60 @@
 from datetime import datetime
 import sqlalchemy.orm
 
+import json
+
 from .schema import Base
 from .schema import Apikeys
 
 class functions():
-    def retrieve_api_keys(session: sqlalchemy.orm.Session) -> dict:
+    def retrieve_api_keys(session: sqlalchemy.orm.Session) -> tuple[list[dict], list[dict]]:
+        """ AI GENERATED
+        Retrieve all API keys from the database and separate them into valid and expired keys.
+
+        Parameters
+        ----------
+        session : sqlalchemy.orm.Session
+            An active SQLAlchemy session connected to the database.
+
+        Returns
+        -------
+        tuple[list[dict], list[dict]]
+            A tuple containing two lists of dictionaries:
+            - The first list contains valid API keys (not expired at the time of the call).
+            - The second list contains expired API keys (already expired at the time of the call).
+
+            Each dictionary has the following keys:
+            - 'id': int, the primary key of the API key record
+            - 'key': str, the API key string
+            - 'creation': str, timestamp of when the key was created
+            - 'expiration': str or None, timestamp of when the key expires, or None if no expiration
+
+        Notes
+        -----
+        - Keys with no expiration date are considered valid.
+        - Expiration comparison is performed against the current system time.
+        """
+
         time_now = datetime.now()
+        stored_keys = session.query(Apikeys).all()
 
-        api_keys = session.query(Apikeys).all()
-        print(dict(api_keys))
+        expired_keys = []
+        valid_keys = []
 
-        return {}
+        for key in stored_keys:
+            expiration_date = None
+
+            if key.key_expiration:
+                expiration_date = datetime.strptime(key.key_expiration, "%Y-%m-%d %H:%M:%S")
+
+            key_dict = key.to_dict()
+
+            if expiration_date is None or expiration_date > time_now:
+                valid_keys.append(key_dict)
+            else:
+                expired_keys.append(key_dict)
+        
+        print("VALID:", json.dumps(valid_keys,indent=4))
+        print("INVALID", json.dumps(expired_keys, indent=4))
+
+        return valid_keys, expired_keys

src/modules/database/schema.py

@@ -24,11 +24,19 @@ class User(Base):
 class Apikeys(Base):
     __tablename__ = 'apikeys'
 
-    apikey_id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, index=True)
-    apikey_data = sqlalchemy.Column(sqlalchemy.String)
+    key_id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, index=True)
+    key_data = sqlalchemy.Column(sqlalchemy.String)
 
-    apikey_creation = sqlalchemy.Column(sqlalchemy.String)
-    apikey_expiration = sqlalchemy.Column(sqlalchemy.String)
+    key_creation = sqlalchemy.Column(sqlalchemy.String)
+    key_expiration = sqlalchemy.Column(sqlalchemy.String)
+
+    def to_dict(self):
+        return {
+            'id': self.key_id,
+            'key': self.key_data,
+            'creation': self.key_creation,
+            'expiration': self.key_expiration,
+        }
 
 class Wireguard(Base):
     __tablename__ = 'wireguard_interfaces'

src/modules/routes/response.py

@@ -1,12 +1,15 @@
 #!/bin/env python3
 
 import flask
+import json
 
-def make_resp_obj(success: bool = True, message: str = "", data: dict = {}, http_code: int = 200) -> flask.wrappers.Response:
-    response = flask.make_response({
-        "message": message,
-        "status": success,
-        "data": data
-    }, http_code)
+def make_resp_obj(message: str = "", data: dict = {}, http_code: int = 200) -> flask.wrappers.Response:
+    resp_json_data = json.dumps({
+        'message': message,
+        'data': data
+    })
+    
+    response = flask.make_response(resp_json_data, http_code)
+    response.mimetype = 'application/json'
 
     return response

src/modules/routes/routes.py

@@ -1,57 +1,101 @@
 #!/bin/env python3
 
+import logging as log
+
 import flask
 
+import json
+
 from .response import make_resp_obj
+from .utilities import helpers
+
 from ..database.functions import functions
-from ..utilities.utilities import utilities
+from ..utilities.utilities import utilities as util
 
 routes = flask.Blueprint("routes", __name__)
 
+white_list = [
+    "/client",
+    "/static/",
+    "/fileDownload",
+    "authenticate",
+    "getDashboardConfiguration",
+    "getDashboardTheme",
+    "getDashboardVersion",
+    "sharePeer/get",
+    "isTotpEnabled",
+    "locale",
+    "validateAuthentication",
+]
+
 @routes.before_request
 def auth_required():
     if flask.request.method.lower() == "options":
-        return make_resp_obj(True, "", flask.jsonify({"status": True}), 200)
+        return make_resp_obj("", {"status": True}, 200)
 
-    ok, config_server = utilities.filter_config(flask.current_app.wgd_config, 'SERVER')
+    ok, config_server = util.filter_config(flask.current_app.wgd_config, 'SERVER')
     if not ok:
-        return make_resp_obj(False, "Internal Error", {}, 500)
+        return make_resp_obj("Internal error", {}, 500)
 
-    auth_required = config_server.get('auth_req', True) # Set to true for a safe default
+    auth_required_flag = config_server.get('auth_req', True)
+    api_key_enabled = config_server.get("wgdashboard_apikey", False)
 
-    if not auth_required:
+    if not auth_required_flag:
         return
 
-    whiteList = [
-        '/client',
-        '/static/',
-        '/fileDownload',
-        'validateAuthentication',
-        'authenticate',
-        'getDashboardConfiguration',
-        'getDashboardTheme',
-        'getDashboardVersion',
-        'sharePeer/get',
-        'isTotpEnabled',
-        'locale'
-    ]
+    path = flask.request.path
+    api_key = flask.request.headers.get("wgdashboard-apikey")
 
-    request_path = flask.request.path
-    http_headers = flask.request.headers
-    api_key = http_headers.get("wgdashboard-apikey")
+    if api_key and api_key_enabled:
+        if helpers.is_valid_api_key(api_key):
+            return
+        else:
+            return make_resp_obj("WGDashboard API-key does not exist or is invalid/expired", {}, 401)
 
-    api_key_enabled = config_server.get("wgdashboard_apikey", False)
-    registered_api_keys = functions.retrieve_api_keys(flask.current_app.db_session)
+    if flask.session.get("role") == "admin":
+        return
 
-    if not api_key:
-        response = make_resp_obj()
-        return response
+    if helpers.is_path_allowed(path, white_list, flask.session):
+        return
 
-@routes.route("/")
+    return make_resp_obj("Unauthorized access", {}, 401)
+
+@routes.route('/api/authenticate', methods=["POST"])
+def api_authenticate():
+    ok, config_server = util.filter_config(flask.current_app.wgd_config, 'SERVER')
+    if not ok:
+        return make_resp_obj("Internal error", {}, 500)
+
+    auth_required_flag = config_server.get('auth_req', True)
+
+    if not auth_required_flag:
+        _, config_other = util.filter_config(flask.current_app.wgd_config, 'OTHER')
+
+        return make_resp_obj(
+            "Login successful, no authentication required",
+            {"welcome_session": config_other.get("welcome_session", False)},
+            200
+        )
+
+    data = request.get_json()
+    if not data:
+        return make_resp_obj("Invalid request body", {}, 400)
+    
+    return make_resp_obj("Authentication required", {}, 401)
+
+@routes.route('/', methods=["GET"])
 def index():
-    return make_resp_obj(True, "/ Endpoint", flask.jsonify({"message": "Hello from routes file!"}), 200)
+    return make_resp_obj(
+        "Pong from the /",
+        {},
+        200
+    )
 
-@routes.route("/health")
-@routes.route("/healthz")
+@routes.route('/health', methods=["GET"])
+@routes.route('/healthz', methods=["GET"])
 def health():
-    return make_resp_obj(True, "Health Endpoint", flask.jsonify({"status": "ok"}), 200)
+    return make_resp_obj(
+        "Health Endpoint",
+        {"status": "ok"},
+        200
+    )

src/modules/routes/utilities.py

@@ -0,0 +1,27 @@
+#!/bin/env python3
+
+import flask
+
+from ..database.functions import functions
+
+class helpers():
+    @staticmethod
+    def is_valid_api_key(api_key: str) -> bool:
+        valid_keys, _ = functions.retrieve_api_keys(flask.current_app.db_session)
+        valid_key_strings = []
+
+        for key in valid_keys:
+            valid_key_strings.append(key["key"])
+
+        return api_key in valid_key_strings
+
+    @staticmethod
+    def is_path_allowed(path: str, white_list: list[str], session_data: dict) -> bool:
+        # Allow if session has admin role
+        if session_data.get("role") == "admin" and "username" in session_data:
+            return True
+        # Check white list
+        for p in white_list:
+            if p in path:
+                return True
+        return False

src/modules/utilities/utilities.py

@@ -10,12 +10,10 @@ class utilities():
         '''
         Helper function to grab a specific part of the config
         '''
-
-        log.debug(f'searching for section: {filter_keyword}')
-        for section in config_contents:
-            if str(section).lower() == filter_keyword.lower():
-                return True, dict(config_contents[section].items())
-
+        for section_name, section_values in config_contents.items():
+            if str(section_name).lower() == filter_keyword.lower():
+                if isinstance(section_values, dict):
+                    return True, dict(section_values)
         return False, {}
 
     @staticmethod
@@ -24,7 +22,6 @@ class utilities():
         Make the directory if it does not exist yet, return only true if the directory was missing and created.
         '''
 
-        log.debug(f'checking if the directory at: {path} exists')
         if os.path.exists(path) and os.path.isdir(path):
             return True