config for oauth

api/__init__.py

@@ -1,22 +1,61 @@
 #api/__init__.py
 import os
 from functools import wraps
-from flask import jsonify, session, Blueprint
+from flask import jsonify, session, Blueprint, request, g
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
-
+from jwt import decode, ExpiredSignatureError, InvalidTokenError
 import importlib
+import requests
+from threading import Lock
 
+_public_key_cache = None
+_lock = Lock()
+
+def keycloak_public_key():
+    global _public_key_cache
+    if _public_key_cache:
+        return _public_key_cache
+    with _lock:
+        if _public_key_cache:
+            return _public_key_cache
+
+        url = "https://login.hangman-lab.top/realms/Hangman-Lab/protocol/openid-connect/certs"
+        response = requests.get(url)
+        jwks = response.json()
+        public_key = jwks["keys"][0]["x5c"][0]
+        _public_key_cache = f"-----BEGIN CERTIFICATE-----\n{public_key}\n-----END CERTIFICATE-----"
+        return _public_key_cache
+
+def verify_token(token):
+    try:
+        decoded = decode(
+            token,
+            keycloak_public_key(),
+            algorithms=["RS256"],
+            audience="labdev"
+        )
+        return decoded
+    except ExpiredSignatureError:
+        return None
+    except InvalidTokenError:
+        return None
 
 def require_auth(roles=[]):
     def decorator(func):
         @wraps(func)
         def wrapper(*args, **kwargs):
-            user = session.get('user')
+            auth_header = request.headers.get('Authorization')
-            if not user:
+            if not auth_header or not auth_header.startswith('Bearer'):
                 return jsonify({"error": "Unauthorized"}), 401
-            if user.get('role') not in roles:
+            token = auth_header.split(" ")[1]
+            decoded = verify_token(token)
+            if not decoded:
+                return jsonify({"error": "Invalid or expired token"}), 401
+            user_roles = decoded.get("roles", [])
+            if roles and not set(roles).issubset(set(user_roles)):
                 return jsonify({"error": "Forbidden, permission denied"}), 403
+            g.user = decoded
             return func(*args, **kwargs)
         return wrapper
     return decorator

api/auth.py

@@ -1,54 +0,0 @@
-#api/auth.py
-from flask import Blueprint, session, redirect, url_for, jsonify
-from authlib.integrations.flask_client import OAuth
-from contexts.RequestContext import RequestContext
-from api import limiter
-import env_provider
-import logging
-
-logger = logging.getLogger(__name__)
-auth_bp = Blueprint('auth', __name__, url_prefix='/api')
-
-oauth = OAuth()
-keycloak = oauth.register(
-    'keycloak',
-    client_id=env_provider.CLIENT_ID,
-    client_secret=env_provider.CLIENT_SECRET,
-    server_metadata_url="https://login.hangman-lab.top/auth/realms/Hangman-Lab/.well-known/openid-configuration",
-    client_kwargs={"scope": "openid email profile"},
-)
-
-@auth_bp.route('/login', methods=['GET'])
-@limiter.limit("20 per minute")
-def login():
-    redirect_uri = url_for("auth.authorize", _external=True)
-    return keycloak.authorize_redirect(redirect_uri)
-
-@auth_bp.route('/authorize', methods=['GET'])
-@limiter.limit("20 per minute")
-def authorize():
-    try:
-        token = keycloak.authorize_access_token()
-        user_info = keycloak.parse_id_token(token)
-        session['user'] = user_info
-        return jsonify({"message": "login successful", "user": user_info})
-    except Exception as e:
-        logger.error(f"Authorization failed: {str(e)}")
-        errno = RequestContext.get_error_id()
-        return jsonify({"error": f"Authorization failed - {errno}"}), 401
-@auth_bp.route('/logout', methods=['GET'])
-def logout():
-    u = session.pop('user', None)
-    if u:
-        logger.info(f"Logged out user: {u}")
-    logout_url = "https://login.hangman-lab.top/auth/realms/Hangman-Lab/protocol/openid-connect/logout"
-    return redirect(logout_url)
-
-@auth_bp.route("/user", methods=["GET"])
-@limiter.limit("80 per minute")
-def user():
-    u = session.get('user')
-    if not u:
-        return jsonify({"username": "guest", "role": "guest"})
-    return jsonify(u)
-

api/config.py

@@ -0,0 +1,18 @@
+from flask import Blueprint, jsonify
+
+import env_provider
+from api import limiter
+config_bp = Blueprint('config', __name__, url_prefix='/api/config')
+
+@config_bp.route('/server_host', methods=['GET'])
+@limiter.limit("120 per minute")
+def server_host():
+    print(env_provider.SERVER_HOST)
+    return jsonify({"value": env_provider.SERVER_HOST}), 200
+
+@config_bp.route('/kc_client_id', methods=['GET'])
+@limiter.limit("120 per minute")
+def kc_client_id():
+    print(env_provider.APP_CLIENT_ID)
+    return jsonify({"value": env_provider.APP_CLIENT_ID}), 200
+

api/markdown.py

@@ -16,7 +16,7 @@ markdown_bp = Blueprint('markdown', __name__, url_prefix='/api/markdown')
 def get_markdowns():
     with get_db() as db:
         mds = db.query(Markdown).all()
-    return jsonify([md.to_dict() for md in mds])
+    return jsonify([md.to_dict() for md in mds]), 200
 
 
 

app.py

@@ -45,6 +45,4 @@ def log_request():
 
 if __name__ == '__main__':
     #logger.info("Starting app")
-    with app.app_context():
-        print(url_for("auth.authorize", _external=True))
     app.run(host='0.0.0.0', port=5000)

env_provider.py

@@ -4,7 +4,7 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-CLIENT_ID = os.getenv("CLIENT_ID")
+APP_CLIENT_ID = os.getenv("APP_CLIENT_ID")
 CLIENT_SECRET = os.getenv("CLIENT_SECRET")
 
 DB_HOST = os.getenv("DB_HOST")
@@ -12,4 +12,7 @@ DB_PORT = os.getenv("DB_PORT")
 DB_NAME = os.getenv("DB_NAME")
 DB_USER = os.getenv("DB_USER")
 DB_PASSWORD = os.getenv("DB_PASSWORD")
-SESSION_SECRET_KEY = os.getenv("SESSION_SECRET_KEY")
+SESSION_SECRET_KEY = os.getenv("SESSION_SECRET_KEY")
+
+
+SERVER_HOST = os.getenv("SERVER_HOST")

requirements.txt

@@ -22,6 +22,7 @@ ordered-set==4.1.0
 packaging==24.2
 pycparser==2.22
 Pygments==2.18.0
+PyJWT==2.10.1
 PyMySQL==1.1.1
 python-dotenv==1.0.1
 requests==2.32.3