updated auth to allow API keys

Author: Mole1424

auth/auth.py

@@ -1,11 +1,15 @@
 import os
+import secrets
 from functools import wraps
 from typing import Callable
 
 from authlib.integrations.flask_client import OAuth
-from flask import Blueprint, Flask, abort, redirect, session, url_for
+from flask import Blueprint, Flask, abort, jsonify, redirect, request, session, url_for
+from werkzeug.security import check_password_hash, generate_password_hash
 from werkzeug.wrappers import Response
 
+from schema import APIKey, db
+
 oauth = OAuth()
 
 
@@ -86,3 +90,107 @@ def logout() -> Response:
     # if no id token, just clear session and redirect to index
     session.clear()
     return redirect(url_for("index"))
+
+
+def create_api_key(owner: str) -> dict:
+    """Create a new API key"""
+    key = secrets.token_urlsafe(32)
+    api_key = APIKey(generate_password_hash(key), owner.strip().lower())
+    db.session.add(api_key)
+    db.session.commit()
+    return {
+        "id": api_key.id,
+        "key": key,
+        "owner": api_key.owner,
+        "active": api_key.active,
+        "created_at": api_key.created_at.isoformat(),
+    }
+
+
+def get_api_keys() -> list[dict]:
+    """Get all API keys"""
+    api_keys = APIKey.query.all()
+    return [key.to_dict() for key in api_keys]
+
+
+def get_api_key(key_id: int) -> dict | None:
+    """Get an API key by its ID"""
+    api_key = APIKey.query.get(key_id)
+    if not api_key:
+        return None
+    return api_key.to_dict()
+
+
+def disable_api_key(key_id: int) -> dict | None:
+    """Disable an API key by its ID"""
+    api_key = APIKey.query.get(key_id)
+    if not api_key:
+        return None
+    api_key.deactivate()
+    return api_key.to_dict()
+
+
+def is_valid_api_key(api_key: str) -> bool:
+    """Check if an API key is valid"""
+    for key in APIKey.query.filter_by(active=True).all():
+        if check_password_hash(key.key, api_key):
+            return True
+    return False
+
+
+def valid_api_auth(f: Callable) -> Callable:
+    """decorater to check if valid API auth"""
+
+    @wraps(f)
+    def decorated_function(*args: object, **kwargs: object) -> Response:
+        # check if exec
+        if is_exec():
+            return f(*args, **kwargs)
+        # check if valid API key
+        # american spelling for convention
+        api_key = request.headers.get("Authorization")
+        if not api_key or not is_valid_api_key(api_key):
+            return abort(403, "Invalid API key or not authorised.")
+        return f(*args, **kwargs)
+
+    return decorated_function
+
+
+auth_api_bp = Blueprint("auth_api", __name__, url_prefix="/api/auth")
+
+
+@auth_api_bp.route("/create", methods=["POST"])
+@is_exec_wrapper
+def create_api_key_api() -> tuple[Response, int]:
+    """Create a new API key"""
+    data = request.get_json()
+    if not data or "owner" not in data:
+        return jsonify({"error": "Owner is required"}), 400
+    return jsonify(create_api_key(data["owner"])), 201
+
+
+@auth_api_bp.route("/<int:key_id>", methods=["GET"])
+@is_exec_wrapper
+def get_api_key_api(key_id: int) -> tuple[Response, int]:
+    """Get an API key by its ID"""
+    api_key = get_api_key(key_id)
+    if not api_key:
+        return jsonify({"error": "API key not found"}), 404
+    return jsonify(api_key), 200
+
+
+@auth_api_bp.route("/disable/<int:key_id>", methods=["POST"])
+@is_exec_wrapper
+def disable_api_key_api(key_id: int) -> tuple[Response, int]:
+    """Disable an API key by its ID"""
+    api_key = disable_api_key(key_id)
+    if not api_key:
+        return jsonify({"error": "API key not found"}), 404
+    return jsonify(api_key), 200
+
+
+@auth_api_bp.route("/keys", methods=["GET"])
+@is_exec_wrapper
+def get_api_keys_api() -> tuple[Response, int]:
+    """Get all API keys"""
+    return jsonify(get_api_keys()), 200

events/api.py

@@ -5,9 +5,9 @@
 import pytz
 import requests
 from flask import Blueprint, Response, jsonify, request
-from schema import Event, Tag, Week, db
 
-from auth.auth import is_exec_wrapper
+from auth.auth import valid_api_auth
+from schema import Event, Tag, Week, db
 
 # bind endpoints to /api/events/...
 events_api_bp = Blueprint("events", __name__, url_prefix="/api/events")
@@ -108,7 +108,7 @@ def get_datetime_from_string(date_str: str) -> datetime | str:
 
 
 @events_api_bp.route("/create", methods=["POST"])
-@is_exec_wrapper
+@valid_api_auth
 def create_event_api() -> tuple[Response, int]:  # noqa: PLR0911
     """Create a new event"""
     data = request.get_json()
@@ -295,7 +295,7 @@ def get_week_from_date(date: datetime) -> Week | None:  # noqa: PLR0912
 
 
 @events_api_bp.route("/create_repeat", methods=["POST"])
-@is_exec_wrapper
+@valid_api_auth
 def create_repeat_event_api() -> tuple[Response, int]:  # noqa: PLR0911
     """Create a bunch of events at once"""
     data = request.get_json()
@@ -420,7 +420,7 @@ def get_week_by_date(date_str: str) -> tuple[Response, int]:
 
 
 @events_api_bp.route("/<int:event_id>", methods=["PATCH"])
-@is_exec_wrapper
+@valid_api_auth
 def edit_event(event_id: int) -> tuple[Response, int]:  # noqa: PLR0911, PLR0912
     """Edit an existing event"""
     event = Event.query.get(event_id)
@@ -488,7 +488,7 @@ def edit_event(event_id: int) -> tuple[Response, int]:  # noqa: PLR0911, PLR0912
 
 
 @events_api_bp.route("/<int:event_id>", methods=["DELETE"])
-@is_exec_wrapper
+@valid_api_auth
 def delete_event(event_id: int) -> tuple[Response, int]:
     """Delete an existing event"""
     event = Event.query.get(event_id)

fulcrum.py

@@ -6,6 +6,7 @@
 
 from auth.auth import auth_bp, configure_oauth, is_exec, is_logged_in
 from events.api import events_api_bp
+from schema import initialise_db
 
 # if .env file exists, load it
 if Path(".env").exists():
@@ -15,6 +16,9 @@
 app = Flask(__name__)
 app.secret_key = os.getenv("SECRET_KEY")
 
+# initialise database
+initialise_db(app)
+
 # setup oauth and add routes
 configure_oauth(app)
 app.register_blueprint(auth_bp)

events/schema.py schema.pyevents/schema.py renamed to schema.py

@@ -16,6 +16,10 @@ def initialise_db(app: Flask) -> None:
 
     db.init_app(app)
 
+    # create the database tables if they don't exist
+    with app.app_context():
+        db.create_all()
+
 
 class Event(db.Model):
     """Model for an event"""
@@ -173,3 +177,38 @@ def __repr__(self) -> str:
     db.Column("event_id", db.Integer, db.ForeignKey("events.id"), primary_key=True),
     db.Column("tag_id", db.Integer, db.ForeignKey("tags.id"), primary_key=True),
 )
+
+
+class APIKey(db.Model):
+    """Model for an API key"""
+
+    __tablename__ = "api_keys"
+
+    id = db.Column(db.Integer, primary_key=True)
+    key = db.Column(db.String, unique=True, nullable=False)
+    owner = db.Column(db.String, nullable=False)
+    created_at = db.Column(db.DateTime, nullable=False)
+    active = db.Column(db.Boolean, default=True)
+
+    def __init__(self, key: str, owner: str) -> None:
+        self.key = key
+        self.owner = owner
+        self.created_at = datetime.datetime.now(pytz.timezone("Europe/London"))
+        self.active = True
+
+    def __repr__(self) -> str:
+        return f"<APIKey {self.key} (ID: {self.id}) owned by {self.owner}>"
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "key": self.key,
+            "owner": self.owner,
+            "created_at": self.created_at.isoformat(),
+            "active": self.active,
+        }
+
+    def deactivate(self) -> None:
+        """Deactivate the API key"""
+        self.active = False
+        db.session.commit()