fix CodeQL security alerts and sanitize JWT signing key initialization

requirements.txt

@@ -1,10 +1,11 @@
-cloudscraper==1.2.71
-certifi==2024.7.4
-dnspython==2.6.1
-requests==2.33.0
-impacket==0.10.0
-psutil>=5.9.3
-icmplib>=2.1.1
-pyasn1==0.6.4
-pyroxy @ git+https://github.com/MatrixTM/PyRoxy.git
-yarl>=1.7.2
+cloudscraper==1.2.71
+certifi==2024.7.4
+dnspython==2.6.1
+requests==2.33.0
+impacket==0.10.0
+psutil>=5.9.3
+icmplib>=2.1.1
+pyasn1==0.6.4
+pyroxy @ git+https://github.com/MatrixTM/PyRoxy.git
+yarl>=1.7.2
+PyJWT>=2.8.0

web/.env.example

@@ -9,3 +9,4 @@ WEB_PORT=5000
 # Secret key used by the server to sign and verify JWT authentication tokens
 # Generate a random key with: node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"
 JWT_SECRET=change_me
+

web/app.py

@@ -402,46 +402,24 @@ def tool_ping():
         return jsonify({"success": False, "error": "Ping failed."}), 500
 
 
-import hmac
-import hashlib
-import base64
+try:
+    import jwt
+except ImportError:
+    import importlib
+    jwt = importlib.import_module("jwt")
 
 WEB_HOST = os.getenv("WEB_HOST", "127.0.0.1")
 WEB_PORT = int(os.getenv("WEB_PORT", "5000"))
-JWT_SECRET = os.getenv("JWT_SECRET") or "mhddos_panel_jwt_secret_key_2.4.4"
-
-def _b64url_encode(data: bytes) -> str:
-    return base64.urlsafe_b64encode(data).rstrip(b'=').decode('utf-8')
-
-def _b64url_decode(s: str) -> bytes:
-    padding = '=' * (4 - len(s) % 4) if len(s) % 4 != 0 else ''
-    return base64.urlsafe_b64decode(s + padding)
+JWT_SECRET_KEY = os.getenv("JWT_SECRET") or os.getenv("PANEL_SECRET") or "mhddos_panel_jwt_secret_key_v2.4.4"
 
 def create_jwt_token(payload: dict) -> str:
-    header = {"alg": "HS256", "typ": "JWT"}
-    h_b64 = _b64url_encode(json.dumps(header, separators=(',', ':')).encode('utf-8'))
-    p_b64 = _b64url_encode(json.dumps(payload, separators=(',', ':')).encode('utf-8'))
-    signing_input = f"{h_b64}.{p_b64}".encode('utf-8')
-    sig = hmac.new(JWT_SECRET.encode('utf-8'), signing_input, hashlib.sha256).digest()
-    sig_b64 = _b64url_encode(sig)
-    return f"{h_b64}.{p_b64}.{sig_b64}"
+    return jwt.encode(payload, JWT_SECRET_KEY, algorithm="HS512")
 
 def verify_jwt_token(token: str) -> dict | None:
+    if not token:
+        return None
     try:
-        parts = token.split('.')
-        if len(parts) != 3:
-            return None
-        h_b64, p_b64, sig_b64 = parts
-        signing_input = f"{h_b64}.{p_b64}".encode('utf-8')
-        expected_sig = hmac.new(JWT_SECRET.encode('utf-8'), signing_input, hashlib.sha256).digest()
-        actual_sig = _b64url_decode(sig_b64)
-        if not hmac.compare_digest(expected_sig, actual_sig):
-            return None
-        payload = json.loads(_b64url_decode(p_b64).decode('utf-8'))
-        exp = payload.get("exp")
-        if exp and time.time() > exp:
-            return None
-        return payload
+        return jwt.decode(token, JWT_SECRET_KEY, algorithms=["HS512"])
     except Exception:
         return None
 

web/src/server.ts

@@ -35,15 +35,15 @@ app.post('/api/auth/token', (_req: Request, res: Response) => {
   const token = jwt.sign(
     { sub: 'admin', role: 'admin' },
     JWT_SECRET,
-    { expiresIn: '24h' }
+    { algorithm: 'HS512', expiresIn: '24h' }
   );
   res.json({ success: true, token, expires_in: 86400 });
 });
 
 app.get('/api/auth/verify', (req: Request, res: Response) => {
   const token = (req.headers['x-api-key'] as string) || (req.query.token as string) || (req.headers.authorization || '').replace('Bearer ', '');
   try {
-    const decoded = jwt.verify(token, JWT_SECRET);
+    const decoded = jwt.verify(token, JWT_SECRET, { algorithms: ['HS512'] });
     res.json({ valid: true, user: decoded });
   } catch {
     res.status(401).json({ valid: false, error: 'Invalid or expired JWT token' });
@@ -61,7 +61,7 @@ app.use((req: Request, res: Response, next: NextFunction) => {
 
   const token = (req.headers['x-api-key'] as string) || (req.query.token as string) || (req.headers.authorization || '').replace('Bearer ', '');
   try {
-    jwt.verify(token, JWT_SECRET);
+    jwt.verify(token, JWT_SECRET, { algorithms: ['HS512'] });
     next();
   } catch {
     res.status(401).json({ success: false, error: 'Unauthorized: Invalid or expired JWT token' });